litestar¶
API Reference for the Litestar
extensions module
Note
Private methods and attributes are not included in the API reference.
- class advanced_alchemy.extensions.litestar.AlembicAsyncConfig[source]¶
Bases:
GenericAlembicConfig
Configuration for an Async Alembic’s Config class.
- __init__(script_config: str = 'alembic.ini', version_table_name: str = 'alembic_versions', version_table_schema: str | None = None, script_location: str = 'migrations', user_module_prefix: str | None = 'sa.', render_as_batch: bool = True, compare_type: bool = False, template_path: str = '/home/runner/work/advanced-alchemy/advanced-alchemy/advanced_alchemy/alembic/templates') None ¶
- class advanced_alchemy.extensions.litestar.AlembicSyncConfig[source]¶
Bases:
GenericAlembicConfig
Configuration for Alembic’s synchronous migrations.
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', user_module_prefix: str | None = 'sa.', render_as_batch: bool = True, compare_type: bool = False, template_path: str = '/home/runner/work/advanced-alchemy/advanced-alchemy/advanced_alchemy/alembic/templates') None ¶
- class advanced_alchemy.extensions.litestar.AsyncSessionConfig[source]¶
Bases:
GenericSessionConfig
[AsyncConnection
,AsyncEngine
,AsyncSession
]SQLAlchemy async session config.
- __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 advanced_alchemy.extensions.litestar.EngineConfig[source]¶
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 ¶
- class advanced_alchemy.extensions.litestar.SQLAlchemyAsyncConfig[source]¶
Bases:
SQLAlchemyAsyncConfig
Litestar 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 applicationState
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.
- set_default_exception_handler: bool = True¶
Sets the default exception handler on application start.
- __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, bind_key: str | None = None, 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', set_default_exception_handler: bool = True) None ¶
- create_session_maker() Callable[[], AsyncSession] [source]¶
Get a session maker. If none exists yet, create one.
- Returns:
Session factory used by the plugin.
- provide_engine(state: State) AsyncEngine [source]¶
Create an engine instance.
- Parameters:
state¶ – The
Litestar.state
instance.- Returns:
An engine 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 [source]¶
Create all metadata
- Parameters:
app¶ (Litestar) – The
Litestar
instance
- class advanced_alchemy.extensions.litestar.SQLAlchemyDTO[source]¶
Bases:
AbstractDTO
,Generic
[T
]Support for domain modelling with SQLAlchemy.
- config: ClassVar[SQLAlchemyDTOConfig]¶
Config objects to define properties of the DTO.
- model_type: type[T]¶
If
annotation
is an iterable, this is the inner type, otherwise will be the same asannotation
.
- classmethod generate_field_definitions(model_type: type[sqlalchemy.orm.DeclarativeBase]) Generator[DTOFieldDefinition, None, None] [source]¶
Generate DTO field definitions from a SQLAlchemy model.
- Parameters:
model_type¶ (Type[sqlalchemy.orm.DeclarativeBase]) – The SQLAlchemy model type to generate field definitions from.
- Yields:
collections.abc.Generator[litestar.dto.data_structures.DTOFieldDefinition, None, None] – A generator yielding DTO field definitions.
- Raises:
RuntimeError – If the mapper cannot be found for the model type.
NotImplementedError – If an unsupported property or extension type is encountered.
ImproperConfigurationError – If a type cannot be parsed from an element.
- classmethod detect_nested_field(field_definition: FieldDefinition) bool [source]¶
Return
True
iffield_definition
represents a nested model field.- Parameters:
field_definition¶ – inspect type to determine if field represents a nested model.
- Returns:
True
iffield_definition
represents a nested model field.
- class advanced_alchemy.extensions.litestar.SQLAlchemyDTOConfig[source]¶
Bases:
DTOConfig
Additional controls for the generated SQLAlchemy DTO.
- exclude: AbstractSet[str | InstrumentedAttribute[Any]]¶
Explicitly exclude fields from the generated DTO.
If exclude is specified, all fields not specified in exclude will be included by default.
Notes
- The field names are dot-separated paths to nested fields, e.g.
"address.street"
will exclude the
"street"
field from a nested"address"
model.
- The field names are dot-separated paths to nested fields, e.g.
- ‘exclude’ mutually exclusive with ‘include’ - specifying both values will raise an
ImproperlyConfiguredException
.
- include: AbstractSet[str | InstrumentedAttribute[Any]]¶
Explicitly include fields in the generated DTO.
If include is specified, all fields not specified in include will be excluded by default.
Notes
- The field names are dot-separated paths to nested fields, e.g.
"address.street"
will include the
"street"
field from a nested"address"
model.
- The field names are dot-separated paths to nested fields, e.g.
- ‘include’ mutually exclusive with ‘exclude’ - specifying both values will raise an
ImproperlyConfiguredException
.
- 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 | InstrumentedAttribute[Any]] = <factory>, include: AbstractSet[str | InstrumentedAttribute[Any]] = <factory>, rename_fields: dict[str | InstrumentedAttribute[Any], 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 advanced_alchemy.extensions.litestar.SQLAlchemyInitPlugin[source]¶
Bases:
InitPluginProtocol
,CLIPluginProtocol
,SlotsBase
SQLAlchemy application lifecycle configuration.
- __init__(config: SQLAlchemyAsyncConfig | SQLAlchemySyncConfig | Sequence[SQLAlchemyAsyncConfig | SQLAlchemySyncConfig]) None [source]¶
Initialize
SQLAlchemyPlugin
.- Parameters:
config¶ – configure DB connection and hook handlers and dependencies.
- on_cli_init(cli: Group) None [source]¶
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()])
- class advanced_alchemy.extensions.litestar.SQLAlchemyPlugin[source]¶
Bases:
InitPluginProtocol
,SlotsBase
A plugin that provides SQLAlchemy integration.
- __init__(config: SQLAlchemyAsyncConfig | SQLAlchemySyncConfig | list[SQLAlchemyAsyncConfig | SQLAlchemySyncConfig]) None [source]¶
Initialize
SQLAlchemyPlugin
.- Parameters:
config¶ – configure DB connection and hook handlers and dependencies.
- class advanced_alchemy.extensions.litestar.SQLAlchemySerializationPlugin[source]¶
Bases:
SerializationPluginProtocol
,SlotsBase
- class advanced_alchemy.extensions.litestar.SQLAlchemySyncConfig[source]¶
Bases:
SQLAlchemySyncConfig
Litestar 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 applicationState
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.
- set_default_exception_handler: bool = True¶
Sets the default exception handler on application start.
- __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, bind_key: str | None = None, 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', set_default_exception_handler: bool = True) None ¶
- create_session_maker() Callable[[], Session] [source]¶
Get a session maker. If none exists yet, create one.
- Returns:
Session factory used by the plugin.
- provide_engine(state: State) Engine [source]¶
Create an engine instance.
- Parameters:
state¶ – The
Litestar.state
instance.- Returns:
An engine 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 [source]¶
Create all metadata
- Parameters:
app¶ (Litestar) – The
Litestar
instance
- class advanced_alchemy.extensions.litestar.SyncSessionConfig[source]¶
Bases:
GenericSessionConfig
[Connection
,Engine
,Session
]Configuration for synchronous SQLAlchemy sessions.
- __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 ¶
- async advanced_alchemy.extensions.litestar.async_autocommit_before_send_handler(message: Message, scope: Scope) None ¶
Handle commit/rollback, closing and cleaning up sessions before sending.
- advanced_alchemy.extensions.litestar.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_advanced_alchemy.extensions.litestar.async_autocommit_handler_maker.commit_on_redirect: Issue a commit when the response status is a redirect (
3XX
) :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.async_autocommit_handler_maker.extra_commit_statuses: A set of additional status codes that trigger a commit :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.async_autocommit_handler_maker.extra_rollback_statuses: A set of additional status codes that trigger a rollback :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.async_autocommit_handler_maker.session_scope_key: The key to use within the application state- Returns:
The handler callable
- async advanced_alchemy.extensions.litestar.async_default_before_send_handler(message: Message, scope: Scope) None ¶
Handle commit/rollback, closing and cleaning up sessions before sending.
- advanced_alchemy.extensions.litestar.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_advanced_alchemy.extensions.litestar.async_default_handler_maker.session_scope_key: The key to use within the application state
- Returns:
The handler callable
- advanced_alchemy.extensions.litestar.get_database_migration_plugin(app: Litestar) SQLAlchemyInitPlugin [source]¶
Retrieve a database migration plugin from the Litestar application’s plugins.
This function attempts to find and return either the SQLAlchemyPlugin or SQLAlchemyInitPlugin. If neither plugin is found, it raises an ImproperlyConfiguredException.
- advanced_alchemy.extensions.litestar.sync_autocommit_before_send_handler(message: Message, scope: Scope) None ¶
Handle commit/rollback, closing and cleaning up sessions before sending.
- advanced_alchemy.extensions.litestar.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_advanced_alchemy.extensions.litestar.sync_autocommit_handler_maker.commit_on_redirect: Issue a commit when the response status is a redirect (
3XX
) :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.sync_autocommit_handler_maker.extra_commit_statuses: A set of additional status codes that trigger a commit :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.sync_autocommit_handler_maker.extra_rollback_statuses: A set of additional status codes that trigger a rollback :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.sync_autocommit_handler_maker.session_scope_key: The key to use within the application state- Returns:
The handler callable
- advanced_alchemy.extensions.litestar.sync_default_before_send_handler(message: Message, scope: Scope) None ¶
Handle commit/rollback, closing and cleaning up sessions before sending.
- advanced_alchemy.extensions.litestar.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_advanced_alchemy.extensions.litestar.sync_default_handler_maker.session_scope_key: The key to use within the application state
- Returns:
The handler callable