plugins#

class litestar.contrib.sqlalchemy.plugins.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.contrib.sqlalchemy.plugins.EngineConfig#

Bases: object

Configuration for SQLAlchemy’s Engine.

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

connect_args#: alias of Empty

echo#: alias of Empty

echo_pool#: alias of Empty

enable_from_linting#: alias of Empty

execution_options#: alias of Empty

hide_parameters#: alias of Empty

insertmanyvalues_page_size#: alias of Empty

isolation_level#: alias of Empty

json_deserializer(buf)#: 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.

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.

label_length#: alias of Empty

logging_name#: alias of Empty

max_identifier_length#: alias of Empty

max_overflow#: alias of Empty

__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] = <built-in method decode of msgspec.json.Decoder object>, json_serializer: Callable[[Any], str] = <function encode_json>, 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#

module#: alias of Empty

paramstyle#: alias of Empty

pool#: alias of Empty

poolclass#: alias of Empty

pool_logging_name#: alias of Empty

pool_pre_ping#: alias of Empty

pool_size#: alias of Empty

pool_recycle#: alias of Empty

pool_reset_on_return#: alias of Empty

pool_timeout#: alias of Empty

pool_use_lifo#: alias of Empty

plugins#: alias of Empty

query_cache_size#: alias of Empty

use_insertmanyvalues#: alias of Empty

class litestar.contrib.sqlalchemy.plugins.GenericSQLAlchemyConfig#

Bases: Generic[EngineT, SessionT, SessionMakerT]

Common SQLAlchemy Configuration.

create_engine_callable: Callable[[str], EngineT]#: Callable that creates an AsyncEngine instance or instance of its subclass.

session_config: GenericSessionConfig[Any, Any, Any]#: Configuration options for either the async_sessionmaker or sessionmaker.

session_maker_class: type[sessionmaker[Session] | async_sessionmaker[AsyncSession]]#: Sessionmaker class to use.

connection_string: str | None = None#

Database connection string in one of the formats supported by SQLAlchemy.

Notes

For async connections, the connection string must include the correct async prefix. e.g. 'postgresql+asyncpg://...' instead of 'postgresql://', and for sync connections its the opposite.

engine_config: EngineConfig#

Configuration for the SQLAlchemy engine.

The configuration options are documented in the SQLAlchemy documentation.

session_maker: Callable[[], SessionT] | None = None#

Callable that returns a session.

If provided, the plugin will use this rather than instantiate a sessionmaker.

__init__(create_engine_callable: Callable[[str], EngineT], session_config: GenericSessionConfig[Any, Any, Any], session_maker_class: type[sessionmaker[Session] | async_sessionmaker[AsyncSession]], 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) → None#

engine_instance: EngineT | None = None#

Optional engine to use.

If set, the plugin will use the provided instance rather than instantiate an engine.

create_all: bool = False#: If true, all models are automatically created on engine creation.

metadata: MetaData | None = None#

Optional metadata to use.

If set, the plugin will use the provided instance rather than the default metadata.

enable_touch_updated_timestamp_listener: bool = True#

Enable Created/Updated Timestamp event listener.

This is a listener that will update created_at and updated_at columns on record modification. Disable if you plan to bring your own update mechanism for these columns

property engine_config_dict: dict[str, Any]#

Return the engine configuration as a dict.

Returns:: A string keyed dict of config kwargs for the SQLAlchemy get_engine function.

property session_config_dict: dict[str, Any]#

Return the session configuration as a dict.

Returns:: A string keyed dict of config kwargs for the SQLAlchemy sessionmaker class.

get_engine() → EngineT#

Return an engine. If none exists yet, create one.

Returns:: Getter that returns the engine instance used by the plugin.

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

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

Returns:: Session factory used by the plugin.

class litestar.contrib.sqlalchemy.plugins.GenericSessionConfig#

Bases: Generic[ConnectionT, EngineT, SessionT]

SQLAlchemy async session config.

autobegin#: alias of Empty

autoflush#: alias of Empty

bind#: alias of Empty

binds#: alias of Empty

class_#: alias of Empty

expire_on_commit#: alias of Empty

info#: alias of Empty

join_transaction_mode#: alias of Empty

query_cls#: alias of Empty

twophase#: 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'>) → None#

class litestar.contrib.sqlalchemy.plugins.SQLAlchemyAsyncConfig#: Bases: SQLAlchemyAsyncConfig, _CreateEngineMixin[AsyncEngine]

class litestar.contrib.sqlalchemy.plugins.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.contrib.sqlalchemy.plugins.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.contrib.sqlalchemy.plugins.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.contrib.sqlalchemy.plugins.SQLAlchemySyncConfig#: Bases: SQLAlchemySyncConfig, _CreateEngineMixin[Engine]

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