plugin

class advanced_alchemy.extensions.litestar.plugins.EngineConfig[source]

Bases: EngineConfig

Configuration for SQLAlchemy’s Engine.

For details see: https://docs.sqlalchemy.org/en/20/core/engines.html

__init__(connect_args=<class 'advanced_alchemy.utils.dataclass.Empty'>, echo=<class 'advanced_alchemy.utils.dataclass.Empty'>, echo_pool=<class 'advanced_alchemy.utils.dataclass.Empty'>, enable_from_linting=<class 'advanced_alchemy.utils.dataclass.Empty'>, execution_options=<class 'advanced_alchemy.utils.dataclass.Empty'>, hide_parameters=<class 'advanced_alchemy.utils.dataclass.Empty'>, insertmanyvalues_page_size=<class 'advanced_alchemy.utils.dataclass.Empty'>, isolation_level=<class 'advanced_alchemy.utils.dataclass.Empty'>, json_deserializer=<function decode_json>, json_serializer=<function serializer>, label_length=<class 'advanced_alchemy.utils.dataclass.Empty'>, logging_name=<class 'advanced_alchemy.utils.dataclass.Empty'>, max_identifier_length=<class 'advanced_alchemy.utils.dataclass.Empty'>, max_overflow=<class 'advanced_alchemy.utils.dataclass.Empty'>, module=<class 'advanced_alchemy.utils.dataclass.Empty'>, paramstyle=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool=<class 'advanced_alchemy.utils.dataclass.Empty'>, poolclass=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_logging_name=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_pre_ping=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_size=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_recycle=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_reset_on_return=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_timeout=<class 'advanced_alchemy.utils.dataclass.Empty'>, pool_use_lifo=<class 'advanced_alchemy.utils.dataclass.Empty'>, plugins=<class 'advanced_alchemy.utils.dataclass.Empty'>, query_cache_size=<class 'advanced_alchemy.utils.dataclass.Empty'>, use_insertmanyvalues=<class 'advanced_alchemy.utils.dataclass.Empty'>)
json_deserializer(target_type=_EmptyEnum.EMPTY, type_decoders=None, strict=True)

Decode a JSON string/bytes into an object.

Parameters:
  • value (str | bytes) – Value to decode

  • target_type (type[T] | EmptyType) – An optional type to decode the data into

  • type_decoders (TypeDecodersSequence | None) – Optional sequence of type decoders

  • strict (bool) – Whether type coercion rules should be strict. Setting to False enables a wider set of coercion rules from string to non-string types for all values

Return type:

Any

Returns:

An object

Raises:

SerializationException – If error decoding value.

json_serializer()

Serialize JSON field values.

Parameters:

value (Any) – Any json serializable value.

Return type:

str

Returns:

JSON string.

class advanced_alchemy.extensions.litestar.plugins.SQLAlchemyAsyncConfig[source]

Bases: SQLAlchemyAsyncConfig

Litestar Async SQLAlchemy Configuration.

__init__(create_engine_callable=<function create_async_engine>, session_config=<factory>, session_maker_class=<class 'sqlalchemy.ext.asyncio.session.async_sessionmaker'>, connection_string=None, engine_config=<factory>, session_maker=None, engine_instance=None, create_all=False, metadata=None, enable_touch_updated_timestamp_listener=True, bind_key=None, alembic_config=<factory>, before_send_handler=None, engine_dependency_key='db_engine', session_dependency_key='db_session', engine_app_state_key='db_engine', session_maker_app_state_key='session_maker_class', session_scope_key='_sqlalchemy_db_session', set_default_exception_handler=True)
before_send_handler: Union[BeforeMessageSendHookHandler, Literal['autocommit', 'autocommit_include_redirects'], None] = None

Handler to call before the ASGI message is sent.

The handler should handle closing the session stored in the ASGI scope, if it’s still open, and committing and uncommitted data.

async create_all_metadata(app)[source]

Create all metadata

Parameters:

app (Litestar) – The Litestar instance

Return type:

None

create_app_state_items()[source]

Key/value pairs to be stored in application state.

Return type:

dict[str, Any]

create_session_maker()[source]

Get a session maker. If none exists yet, create one.

Returns:

Session factory used by the plugin.

engine_app_state_key: str = 'db_engine'

Key under which to store the SQLAlchemy engine in the application State instance.

engine_dependency_key: str = 'db_engine'

Key to use for the dependency injection of database engines.

provide_engine(state)[source]

Create an engine instance.

Parameters:

state (State) – The Litestar.state instance.

Return type:

sqlalchemy.ext.asyncio.AsyncEngine

Returns:

An engine instance.

provide_session(state, scope)[source]

Create a session instance.

Parameters:
Return type:

AsyncSession

Returns:

A session instance.

session_dependency_key: str = 'db_session'

Key to use for the dependency injection of database sessions.

session_maker_app_state_key: str = 'session_maker_class'

Key under which to store the SQLAlchemy sessionmaker in the application State instance.

session_scope_key: str = '_sqlalchemy_db_session'

Key under which to store the SQLAlchemy scope in the application.

set_default_exception_handler: bool = True

Sets the default exception handler on application start.

property signature_namespace: dict[str, Any]

Return the plugin’s signature namespace.

Returns:

A string keyed dict of names to be added to the namespace for signature forward reference resolution.

update_app_state(app)[source]

Set the app state with engine and session.

Parameters:

app (Litestar) – The Litestar instance.

Return type:

None

engine_config: EngineConfig

Configuration for the SQLAlchemy engine.

The configuration options are documented in the SQLAlchemy documentation.

session_config: AsyncSessionConfig

Configuration options for the async_sessionmaker.

alembic_config: AlembicAsyncConfig

Configuration for the SQLAlchemy Alembic migrations.

The configuration options are documented in the Alembic documentation.

class advanced_alchemy.extensions.litestar.plugins.SQLAlchemyInitPlugin[source]

Bases: InitPluginProtocol, CLIPlugin, SlotsBase

SQLAlchemy application lifecycle configuration.

__init__(config)[source]

Initialize SQLAlchemyPlugin.

Parameters:

config – configure DB connection and hook handlers and dependencies.

on_app_init(app_config)[source]

Configure application for use with SQLAlchemy.

Parameters:

app_config (AppConfig) – The AppConfig instance.

Return type:

AppConfig

on_cli_init(cli)[source]

Called when the CLI is initialized.

This can be used to extend or override existing commands.

Parameters:

cli (Group) – The root click.Group of the Litestar CLI

Return type:

None

Examples

from litestar import Litestar
from litestar.plugins import CLIPluginProtocol
from click import Group


class CLIPlugin(CLIPluginProtocol):
    def on_cli_init(self, cli: Group) -> None:
        @cli.command()
        def is_debug_mode(app: Litestar):
            print(app.debug)


app = Litestar(plugins=[CLIPlugin()])
class advanced_alchemy.extensions.litestar.plugins.SQLAlchemyPlugin[source]

Bases: InitPluginProtocol, SlotsBase

A plugin that provides SQLAlchemy integration.

__init__(config)[source]

Initialize SQLAlchemyPlugin.

Parameters:

config (Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig, Sequence[Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig]]]) – configure DB connection and hook handlers and dependencies.

on_app_init(app_config)[source]

Configure application for use with SQLAlchemy.

Parameters:

app_config (AppConfig) – The AppConfig instance.

Return type:

AppConfig

class advanced_alchemy.extensions.litestar.plugins.SQLAlchemySerializationPlugin[source]

Bases: SerializationPlugin, SlotsBase

__init__()[source]
create_dto_for_type(field_definition)[source]

Given a parsed type, create a DTO class.

Parameters:

field_definition (FieldDefinition) – A parsed type.

Return type:

type[Any]]

Returns:

A DTO class.

supports_type(field_definition)[source]

Given a value of indeterminate type, determine if this value is supported by the plugin.

Parameters:

field_definition (FieldDefinition) – A parsed type.

Return type:

bool

Returns:

Whether the type is supported by the plugin.

class advanced_alchemy.extensions.litestar.plugins.SQLAlchemySyncConfig[source]

Bases: SQLAlchemySyncConfig

Litestar Sync SQLAlchemy Configuration.

__init__(create_engine_callable=<function create_engine>, session_config=<factory>, session_maker_class=<class 'sqlalchemy.orm.session.sessionmaker'>, connection_string=None, engine_config=<factory>, session_maker=None, engine_instance=None, create_all=False, metadata=None, enable_touch_updated_timestamp_listener=True, bind_key=None, alembic_config=<factory>, before_send_handler=None, engine_dependency_key='db_engine', session_dependency_key='db_session', engine_app_state_key='db_engine', session_maker_app_state_key='session_maker_class', session_scope_key='_sqlalchemy_db_session', set_default_exception_handler=True)
before_send_handler: Union[BeforeMessageSendHookHandler, Literal['autocommit', 'autocommit_include_redirects'], None] = None

Handler to call before the ASGI message is sent.

The handler should handle closing the session stored in the ASGI scope, if it’s still open, and committing and uncommitted data.

create_all_metadata(app)[source]

Create all metadata

Parameters:

app (Litestar) – The Litestar instance

Return type:

None

create_app_state_items()[source]

Key/value pairs to be stored in application state.

Return type:

dict[str, typing.Any]

create_session_maker()[source]

Get a session maker. If none exists yet, create one.

Returns:

Session factory used by the plugin.

engine_app_state_key: str = 'db_engine'

Key under which to store the SQLAlchemy engine in the application State instance.

engine_dependency_key: str = 'db_engine'

Key to use for the dependency injection of database engines.

provide_engine(state)[source]

Create an engine instance.

Parameters:

state (State) – The Litestar.state instance.

Return type:

sqlalchemy.engine.Engine

Returns:

An engine instance.

provide_session(state, scope)[source]

Create a session instance.

Parameters:
Return type:

Session

Returns:

A session instance.

session_dependency_key: str = 'db_session'

Key to use for the dependency injection of database sessions.

session_maker_app_state_key: str = 'session_maker_class'

Key under which to store the SQLAlchemy sessionmaker in the application State instance.

session_scope_key: str = '_sqlalchemy_db_session'

Key under which to store the SQLAlchemy scope in the application.

set_default_exception_handler: bool = True

Sets the default exception handler on application start.

property signature_namespace: dict[str, TypeAliasForwardRef('typing.Any')]

Return the plugin’s signature namespace.

Returns:

A string keyed dict of names to be added to the namespace for signature forward reference resolution.

update_app_state(app)[source]

Set the app state with engine and session.

Parameters:

app (Litestar) – The Litestar instance.

Return type:

None

engine_config: EngineConfig

Configuration for the SQLAlchemy engine.

The configuration options are documented in the SQLAlchemy documentation.

session_config: SyncSessionConfig

Configuration options for the sessionmaker.

alembic_config: AlembicSyncConfig

Configuration for the SQLAlchemy Alembic migrations.

The configuration options are documented in the Alembic documentation.