sqlalchemy#

litestar.plugins.sqlalchemy.async_autocommit_handler_maker(commit_on_redirect: bool = False, extra_commit_statuses: set[int] | None = None, extra_rollback_statuses: set[int] | None = None, session_scope_key: str = '_sqlalchemy_db_session') → Callable[[Message, Scope], Coroutine[Any, Any, None]]#

Set up the handler to issue a transaction commit or rollback based on specified status codes :param _sphinx_paramlinks_litestar.plugins.sqlalchemy.async_autocommit_handler_maker.commit_on_redirect: Issue a commit when the response status is a redirect (3XX) :param _sphinx_paramlinks_litestar.plugins.sqlalchemy.async_autocommit_handler_maker.extra_commit_statuses: A set of additional status codes that trigger a commit :param _sphinx_paramlinks_litestar.plugins.sqlalchemy.async_autocommit_handler_maker.extra_rollback_statuses: A set of additional status codes that trigger a rollback :param _sphinx_paramlinks_litestar.plugins.sqlalchemy.async_autocommit_handler_maker.session_scope_key: The key to use within the application state

Returns:: The handler callable

litestar.plugins.sqlalchemy.sync_autocommit_handler_maker(commit_on_redirect: bool = False, extra_commit_statuses: set[int] | None = None, extra_rollback_statuses: set[int] | None = None, session_scope_key: str = '_sqlalchemy_db_session') → Callable[[Message, Scope], None]#

Set up the handler to issue a transaction commit or rollback based on specified status codes :param _sphinx_paramlinks_litestar.plugins.sqlalchemy.sync_autocommit_handler_maker.commit_on_redirect: Issue a commit when the response status is a redirect (3XX) :param _sphinx_paramlinks_litestar.plugins.sqlalchemy.sync_autocommit_handler_maker.extra_commit_statuses: A set of additional status codes that trigger a commit :param _sphinx_paramlinks_litestar.plugins.sqlalchemy.sync_autocommit_handler_maker.extra_rollback_statuses: A set of additional status codes that trigger a rollback :param _sphinx_paramlinks_litestar.plugins.sqlalchemy.sync_autocommit_handler_maker.session_scope_key: The key to use within the application state

Returns:: The handler callable

litestar.plugins.sqlalchemy.async_default_handler_maker(session_scope_key: str = '_sqlalchemy_db_session') → Callable[[Message, Scope], Coroutine[Any, Any, None]]#

Set up the handler to issue a transaction commit or rollback based on specified status codes :param _sphinx_paramlinks_litestar.plugins.sqlalchemy.async_default_handler_maker.session_scope_key: The key to use within the application state

Returns:: The handler callable

litestar.plugins.sqlalchemy.sync_default_handler_maker(session_scope_key: str = '_sqlalchemy_db_session') → Callable[[Message, Scope], None]#

Set up the handler to issue a transaction commit or rollback based on specified status codes :param _sphinx_paramlinks_litestar.plugins.sqlalchemy.sync_default_handler_maker.session_scope_key: The key to use within the application state

Returns:: The handler callable

litestar.plugins.sqlalchemy.sync_autocommit_before_send_handler(message: Message, scope: Scope) → None#

Handle commit/rollback, closing and cleaning up sessions before sending.

Parameters:

message¶ – ASGI-Message
scope¶ – An ASGI-Scope

Returns:

None

async litestar.plugins.sqlalchemy.async_autocommit_before_send_handler(message: Message, scope: Scope) → None#

Handle commit/rollback, closing and cleaning up sessions before sending.

Parameters:

message¶ – ASGI-Message
scope¶ – An ASGI-Scope

Returns:

None

litestar.plugins.sqlalchemy.sync_default_before_send_handler(message: Message, scope: Scope) → None#

Handle commit/rollback, closing and cleaning up sessions before sending.

Parameters:

message¶ – ASGI-Message
scope¶ – An ASGI-Scope

Returns:

None

async litestar.plugins.sqlalchemy.async_default_before_send_handler(message: Message, scope: Scope) → None#

Handle commit/rollback, closing and cleaning up sessions before sending.

Parameters:

message¶ – ASGI-Message
scope¶ – An ASGI-Scope

Returns:

None

class litestar.plugins.sqlalchemy.AlembicAsyncConfig#

Bases: GenericAlembicConfig

Configuration for an Async Alembic’s Config.

For details see: https://alembic.sqlalchemy.org/en/latest/api/config.html

__init__(script_config: str = 'alembic.ini', version_table_name: str = 'alembic_versions', version_table_schema: str | None = None, script_location: str = 'migrations', target_metadata: MetaData = MetaData(), user_module_prefix: str | None = 'sa.', render_as_batch: bool = True, compare_type: bool = False, template_path: str = '/home/runner/work/litestar/litestar/.venv/lib/python3.12/site-packages/advanced_alchemy/alembic/templates') → None#

class litestar.plugins.sqlalchemy.AlembicSyncConfig#

Bases: GenericAlembicConfig

Configuration for a Sync Alembic’s Config.

For details see: https://alembic.sqlalchemy.org/en/latest/api/config.html

__init__(script_config: str = 'alembic.ini', version_table_name: str = 'alembic_versions', version_table_schema: str | None = None, script_location: str = 'migrations', target_metadata: MetaData = MetaData(), user_module_prefix: str | None = 'sa.', render_as_batch: bool = True, compare_type: bool = False, template_path: str = '/home/runner/work/litestar/litestar/.venv/lib/python3.12/site-packages/advanced_alchemy/alembic/templates') → None#

class litestar.plugins.sqlalchemy.AsyncSessionConfig#

Bases: GenericSessionConfig[AsyncConnection, AsyncEngine, AsyncSession]

SQLAlchemy async session config.

sync_session_class#: alias of Empty

__init__(autobegin: bool | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, autoflush: bool | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, bind: EngineT | ConnectionT | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, binds: dict[type[Any] | Mapper | TableClause | str, EngineT | ConnectionT] | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, class_: type[SessionT] | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, expire_on_commit: bool | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, info: dict[str, Any] | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, join_transaction_mode: JoinTransactionMode | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, query_cls: type[Query] | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, twophase: bool | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, sync_session_class: type[Session] | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>) → None#

class litestar.plugins.sqlalchemy.SyncSessionConfig#: Bases: GenericSessionConfig[Connection, Engine, Session]

class litestar.plugins.sqlalchemy.SQLAlchemyDTO#

Bases: AbstractDTO, Generic[T]

Support for domain modelling with SQLAlchemy.

config: ClassVar[SQLAlchemyDTOConfig]#: Config objects to define properties of the DTO.

classmethod generate_field_definitions(model_type: type[DeclarativeBase]) → Generator[DTOFieldDefinition, None, None]#

Generate FieldDefinition instances from model_type.

Yields:: FieldDefinition instances.

classmethod detect_nested_field(field_definition: FieldDefinition) → bool#

Return True if field_definition represents a nested model field.

Parameters:: field_definition¶ – inspect type to determine if field represents a nested model.
Returns:: True if field_definition represents a nested model field.

class litestar.plugins.sqlalchemy.SQLAlchemyDTOConfig#

Bases: DTOConfig

Additional controls for the generated SQLAlchemy DTO.

include_implicit_fields: bool | Literal['hybrid-only'] = True#

Fields that are implicitly mapped are included.

Turning this off will lead to exclude all fields not using Mapped annotation,

When setting this to hybrid-only, all implicitly mapped fields are excluded with the exception for hybrid properties.

__init__(exclude: AbstractSet[str] = <factory>, include: AbstractSet[str] = <factory>, rename_fields: dict[str, str] = <factory>, rename_strategy: RenameStrategy | None = None, max_nested_depth: int = 1, partial: bool = False, underscore_fields_private: bool = True, experimental_codegen_backend: bool | None = None, forbid_unknown_fields: bool = False, include_implicit_fields: bool | Literal['hybrid-only'] = True) → None#

class litestar.plugins.sqlalchemy.SQLAlchemyAsyncConfig#

Bases: SQLAlchemyAsyncConfig

Async SQLAlchemy Configuration.

before_send_handler: BeforeMessageSendHookHandler | None | Literal['autocommit', 'autocommit_include_redirects'] = 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.

engine_dependency_key: str = 'db_engine'#: Key to use for the dependency injection of database engines.

session_dependency_key: str = 'db_session'#: Key to use for the dependency injection of database sessions.

engine_app_state_key: str = 'db_engine'#: Key under which to store the SQLAlchemy engine in the application State instance.

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.

engine_config: EngineConfig#

Configuration for the SQLAlchemy engine.

The configuration options are documented in the SQLAlchemy documentation.

create_session_maker() → Callable[[], AsyncSession]#

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

Returns:: Session factory used by the plugin.

__init__(create_engine_callable: Callable[[str], AsyncEngine] = <function create_async_engine>, session_config: AsyncSessionConfig = <factory>, session_maker_class: type[async_sessionmaker[AsyncSession]] = <class 'sqlalchemy.ext.asyncio.session.async_sessionmaker'>, connection_string: str | None = None, engine_config: EngineConfig = <factory>, session_maker: Callable[[], SessionT] | None = None, engine_instance: EngineT | None = None, create_all: bool = False, metadata: MetaData | None = None, enable_touch_updated_timestamp_listener: bool = True, alembic_config: AlembicAsyncConfig = <factory>, before_send_handler: BeforeMessageSendHookHandler | None | Literal['autocommit', 'autocommit_include_redirects'] = None, engine_dependency_key: str = 'db_engine', session_dependency_key: str = 'db_session', engine_app_state_key: str = 'db_engine', session_maker_app_state_key: str = 'session_maker_class', session_scope_key: str = '_sqlalchemy_db_session') → None#

provide_engine(state: State) → AsyncEngine#

Create an engine instance.

Parameters:: state¶ – The Litestar.state instance.
Returns:: An engine instance.

provide_session(state: State, scope: Scope) → AsyncSession#

Create a session instance.

Parameters:

state¶ – The Litestar.state instance.
scope¶ – The current connection’s scope.

Returns:

A session instance.

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.

async create_all_metadata(app: Litestar) → None#

Create all metadata

Parameters:: app¶ (Litestar) – The Litestar instance

create_app_state_items() → dict[str, Any]#: Key/value pairs to be stored in application state.

update_app_state(app: Litestar) → None#

Set the app state with engine and session.

Parameters:: app¶ – The Litestar instance.

class litestar.plugins.sqlalchemy.SQLAlchemyInitPlugin#

Bases: InitPluginProtocol, CLIPluginProtocol, SlotsBase

SQLAlchemy application lifecycle configuration.

__init__(config: SQLAlchemyAsyncConfig | SQLAlchemySyncConfig | Sequence[SQLAlchemyAsyncConfig | SQLAlchemySyncConfig]) → None#

Initialize SQLAlchemyPlugin.

Parameters:: config¶ – configure DB connection and hook handlers and dependencies.

on_cli_init(cli: Group) → None#

Called when the CLI is initialized.

This can be used to extend or override existing commands.

Parameters:: cli¶ – The root click.Group of the Litestar CLI

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()])

on_app_init(app_config: AppConfig) → AppConfig#

Configure application for use with SQLAlchemy.

Parameters:: app_config¶ – The AppConfig instance.

class litestar.plugins.sqlalchemy.SQLAlchemyPlugin#

Bases: InitPluginProtocol, SlotsBase

A plugin that provides SQLAlchemy integration.

__init__(config: SQLAlchemyAsyncConfig | SQLAlchemySyncConfig | list[advanced_alchemy.extensions.litestar.plugins.init.config.asyncio.SQLAlchemyAsyncConfig | advanced_alchemy.extensions.litestar.plugins.init.config.sync.SQLAlchemySyncConfig]) → None#

Initialize SQLAlchemyPlugin.

Parameters:: config¶ – configure DB connection and hook handlers and dependencies.

on_app_init(app_config: AppConfig) → AppConfig#

Configure application for use with SQLAlchemy.

Parameters:: app_config¶ – The AppConfig instance.

class litestar.plugins.sqlalchemy.SQLAlchemySerializationPlugin#

Bases: SerializationPluginProtocol, SlotsBase

__init__() → None#

supports_type(field_definition: FieldDefinition) → bool#

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

Parameters:: field_definition¶ – A parsed type.
Returns:: Whether the type is supported by the plugin.

create_dto_for_type(field_definition: FieldDefinition) → type[SQLAlchemyDTO[Any]]#

Given a parsed type, create a DTO class.

Parameters:: field_definition¶ – A parsed type.
Returns:: A DTO class.

class litestar.plugins.sqlalchemy.SQLAlchemySyncConfig#

Bases: SQLAlchemySyncConfig

Sync SQLAlchemy Configuration.

before_send_handler: BeforeMessageSendHookHandler | None | Literal['autocommit', 'autocommit_include_redirects'] = 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.

engine_dependency_key: str = 'db_engine'#: Key to use for the dependency injection of database engines.

session_dependency_key: str = 'db_session'#: Key to use for the dependency injection of database sessions.

engine_app_state_key: str = 'db_engine'#: Key under which to store the SQLAlchemy engine in the application State instance.

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.

engine_config: EngineConfig#

Configuration for the SQLAlchemy engine.

The configuration options are documented in the SQLAlchemy documentation.

create_session_maker() → Callable[[], Session]#

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

Returns:: Session factory used by the plugin.

__init__(create_engine_callable: Callable[[str], Engine] = <function create_engine>, session_config: SyncSessionConfig = <factory>, session_maker_class: type[sessionmaker[Session]] = <class 'sqlalchemy.orm.session.sessionmaker'>, connection_string: str | None = None, engine_config: EngineConfig = <factory>, session_maker: Callable[[], SessionT] | None = None, engine_instance: EngineT | None = None, create_all: bool = False, metadata: MetaData | None = None, enable_touch_updated_timestamp_listener: bool = True, alembic_config: AlembicSyncConfig = <factory>, before_send_handler: BeforeMessageSendHookHandler | None | Literal['autocommit', 'autocommit_include_redirects'] = None, engine_dependency_key: str = 'db_engine', session_dependency_key: str = 'db_session', engine_app_state_key: str = 'db_engine', session_maker_app_state_key: str = 'session_maker_class', session_scope_key: str = '_sqlalchemy_db_session') → None#

provide_engine(state: State) → Engine#

Create an engine instance.

Parameters:: state¶ – The Litestar.state instance.
Returns:: An engine instance.

provide_session(state: State, scope: Scope) → Session#

Create a session instance.

Parameters:

state¶ – The Litestar.state instance.
scope¶ – The current connection’s scope.

Returns:

A session instance.

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.

create_all_metadata(app: Litestar) → None#

Create all metadata

Parameters:: app¶ (Litestar) – The Litestar instance

create_app_state_items() → dict[str, Any]#: Key/value pairs to be stored in application state.

update_app_state(app: Litestar) → None#

Set the app state with engine and session.

Parameters:: app¶ – The Litestar instance.

class litestar.plugins.sqlalchemy.EngineConfig#

Bases: EngineConfig

Configuration for SQLAlchemy’s Engine.

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

json_deserializer(target_type: type[T] | EmptyType = _EmptyEnum.EMPTY, type_decoders: TypeDecodersSequence | None = None, strict: bool = True) → Any#: For dialects that support the JSON datatype, this is a Python callable that will convert a JSON string to a Python object. By default, this is set to Litestar’s decode_json() function.

__init__(connect_args: dict[Any, Any] | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, echo: _EchoFlagType | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, echo_pool: _EchoFlagType | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, enable_from_linting: bool | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, execution_options: Mapping[str, Any] | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, hide_parameters: bool | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, insertmanyvalues_page_size: int | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, isolation_level: IsolationLevel | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, json_deserializer: Callable[[str], Any] = <function decode_json>, json_serializer: Callable[[Any], str] = <function serializer>, label_length: int | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, logging_name: str | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, max_identifier_length: int | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, max_overflow: int | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, module: Any | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, paramstyle: _ParamStyle | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, pool: Pool | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, poolclass: type[Pool] | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, pool_logging_name: str | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, pool_pre_ping: bool | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, pool_size: int | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, pool_recycle: int | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, pool_reset_on_return: Literal['rollback', 'commit'] | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, pool_timeout: int | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, pool_use_lifo: bool | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, plugins: list[str] | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, query_cache_size: int | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, use_insertmanyvalues: bool | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>) → None#

json_serializer() → str#: For dialects that support the JSON datatype, this is a Python callable that will render a given object as JSON. By default, Litestar’s encode_json() is used.