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='alembic.ini', version_table_name='alembic_versions', version_table_schema=None, script_location='migrations', user_module_prefix='sa.', render_as_batch=True, compare_type=False, template_path='/home/runner/work/advanced-alchemy/advanced-alchemy/advanced_alchemy/alembic/templates')
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='alembic.ini', version_table_name='alembic_versions', version_table_schema=None, script_location='migrations', user_module_prefix='sa.', render_as_batch=True, compare_type=False, template_path='/home/runner/work/advanced-alchemy/advanced-alchemy/advanced_alchemy/alembic/templates')
class advanced_alchemy.extensions.litestar.AsyncSessionConfig[source]

Bases: GenericSessionConfig[AsyncConnection, AsyncEngine, AsyncSession]

SQLAlchemy async session config.

sync_session_class

alias of Empty

__init__(autobegin=<class 'advanced_alchemy.utils.dataclass.Empty'>, autoflush=<class 'advanced_alchemy.utils.dataclass.Empty'>, bind=<class 'advanced_alchemy.utils.dataclass.Empty'>, binds=<class 'advanced_alchemy.utils.dataclass.Empty'>, class_=<class 'advanced_alchemy.utils.dataclass.Empty'>, expire_on_commit=<class 'advanced_alchemy.utils.dataclass.Empty'>, info=<class 'advanced_alchemy.utils.dataclass.Empty'>, join_transaction_mode=<class 'advanced_alchemy.utils.dataclass.Empty'>, query_cls=<class 'advanced_alchemy.utils.dataclass.Empty'>, twophase=<class 'advanced_alchemy.utils.dataclass.Empty'>, sync_session_class=<class 'advanced_alchemy.utils.dataclass.Empty'>)
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=_EmptyEnum.EMPTY, type_decoders=None, strict=True)

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.

Return type:

Any

__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_serializer()

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 function is used.

Return type:

str

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 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.

set_default_exception_handler: bool = True

Sets the default exception handler on application start.

__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)
create_session_maker()[source]

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

Return type:

Callable[[], AsyncSession]

Returns:

Session factory used by the plugin.

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.

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

update_app_state(app)[source]

Set the app state with engine and session.

Parameters:

app (Litestar) – The Litestar instance.

Return type:

None

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 as annotation.

classmethod generate_field_definitions(model_type)[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.

Return type:

Generator[DTOFieldDefinition, None, None]

classmethod detect_nested_field(field_definition)[source]

Return True if field_definition represents a nested model field.

Parameters:

field_definition (FieldDefinition) – inspect type to determine if field represents a nested model.

Return type:

bool

Returns:

True if field_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.

  • ‘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.

  • ‘include’ mutually exclusive with ‘exclude’ - specifying both values will raise an

    ImproperlyConfiguredException.

rename_fields: dict[str | InstrumentedAttribute[Any], str]

Mapping of field names, to new name.

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=<factory>, include=<factory>, rename_fields=<factory>, rename_strategy=None, max_nested_depth=1, partial=False, underscore_fields_private=True, experimental_codegen_backend=None, forbid_unknown_fields=False, include_implicit_fields=True)
class advanced_alchemy.extensions.litestar.SQLAlchemyInitPlugin[source]

Bases: InitPluginProtocol, CLIPluginProtocol, SlotsBase

SQLAlchemy application lifecycle configuration.

__init__(config)[source]

Initialize SQLAlchemyPlugin.

Parameters:

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

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()])
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.SQLAlchemyPlugin[source]

Bases: InitPluginProtocol, SlotsBase

A plugin that provides SQLAlchemy integration.

__init__(config)[source]

Initialize SQLAlchemyPlugin.

Parameters:

config (SQLAlchemyAsyncConfig | SQLAlchemySyncConfig | list[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.SQLAlchemySerializationPlugin[source]

Bases: SerializationPluginProtocol, SlotsBase

__init__()[source]
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.

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.

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 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.

set_default_exception_handler: bool = True

Sets the default exception handler on application start.

__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)
create_session_maker()[source]

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

Return type:

Callable[[], Session]

Returns:

Session factory used by the plugin.

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.

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

update_app_state(app)[source]

Set the app state with engine and session.

Parameters:

app (Litestar) – The Litestar instance.

Return type:

None

class advanced_alchemy.extensions.litestar.SyncSessionConfig[source]

Bases: GenericSessionConfig[Connection, Engine, Session]

Configuration for synchronous SQLAlchemy sessions.

__init__(autobegin=<class 'advanced_alchemy.utils.dataclass.Empty'>, autoflush=<class 'advanced_alchemy.utils.dataclass.Empty'>, bind=<class 'advanced_alchemy.utils.dataclass.Empty'>, binds=<class 'advanced_alchemy.utils.dataclass.Empty'>, class_=<class 'advanced_alchemy.utils.dataclass.Empty'>, expire_on_commit=<class 'advanced_alchemy.utils.dataclass.Empty'>, info=<class 'advanced_alchemy.utils.dataclass.Empty'>, join_transaction_mode=<class 'advanced_alchemy.utils.dataclass.Empty'>, query_cls=<class 'advanced_alchemy.utils.dataclass.Empty'>, twophase=<class 'advanced_alchemy.utils.dataclass.Empty'>)
async advanced_alchemy.extensions.litestar.async_autocommit_before_send_handler(message, scope)

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

Parameters:
Return type:

None

Returns:

None

advanced_alchemy.extensions.litestar.async_autocommit_handler_maker(commit_on_redirect=False, extra_commit_statuses=None, extra_rollback_statuses=None, session_scope_key='_sqlalchemy_db_session')

Set up the handler to issue a transaction commit or rollback based on specified status codes :type _sphinx_paramlinks_advanced_alchemy.extensions.litestar.async_autocommit_handler_maker.commit_on_redirect: bool :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) :type _sphinx_paramlinks_advanced_alchemy.extensions.litestar.async_autocommit_handler_maker.extra_commit_statuses: set[int] | None :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.async_autocommit_handler_maker.extra_commit_statuses: A set of additional status codes that trigger a commit :type _sphinx_paramlinks_advanced_alchemy.extensions.litestar.async_autocommit_handler_maker.extra_rollback_statuses: set[int] | None :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.async_autocommit_handler_maker.extra_rollback_statuses: A set of additional status codes that trigger a rollback :type _sphinx_paramlinks_advanced_alchemy.extensions.litestar.async_autocommit_handler_maker.session_scope_key: str :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.async_autocommit_handler_maker.session_scope_key: The key to use within the application state

Return type:

Callable[[Union[HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPServerPushEvent, HTTPDisconnectEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketResponseStartEvent, WebSocketResponseBodyEvent, WebSocketCloseEvent], Union[HTTPScope, WebSocketScope]], Coroutine[Any, Any, None]]

Returns:

The handler callable

async advanced_alchemy.extensions.litestar.async_default_before_send_handler(message, scope)

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

Parameters:
Return type:

None

Returns:

None

advanced_alchemy.extensions.litestar.async_default_handler_maker(session_scope_key='_sqlalchemy_db_session')

Set up the handler to issue a transaction commit or rollback based on specified status codes :type _sphinx_paramlinks_advanced_alchemy.extensions.litestar.async_default_handler_maker.session_scope_key: str :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.async_default_handler_maker.session_scope_key: The key to use within the application state

Return type:

Callable[[Union[HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPServerPushEvent, HTTPDisconnectEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketResponseStartEvent, WebSocketResponseBodyEvent, WebSocketCloseEvent], Union[HTTPScope, WebSocketScope]], Coroutine[Any, Any, None]]

Returns:

The handler callable

advanced_alchemy.extensions.litestar.get_database_migration_plugin(app)[source]

Retrieve a database migration plugin from the Litestar application’s plugins.

Return type:

SQLAlchemyInitPlugin

advanced_alchemy.extensions.litestar.sync_autocommit_before_send_handler(message, scope)

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

Parameters:
Return type:

None

Returns:

None

advanced_alchemy.extensions.litestar.sync_autocommit_handler_maker(commit_on_redirect=False, extra_commit_statuses=None, extra_rollback_statuses=None, session_scope_key='_sqlalchemy_db_session')

Set up the handler to issue a transaction commit or rollback based on specified status codes :type _sphinx_paramlinks_advanced_alchemy.extensions.litestar.sync_autocommit_handler_maker.commit_on_redirect: bool :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) :type _sphinx_paramlinks_advanced_alchemy.extensions.litestar.sync_autocommit_handler_maker.extra_commit_statuses: set[int] | None :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.sync_autocommit_handler_maker.extra_commit_statuses: A set of additional status codes that trigger a commit :type _sphinx_paramlinks_advanced_alchemy.extensions.litestar.sync_autocommit_handler_maker.extra_rollback_statuses: set[int] | None :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.sync_autocommit_handler_maker.extra_rollback_statuses: A set of additional status codes that trigger a rollback :type _sphinx_paramlinks_advanced_alchemy.extensions.litestar.sync_autocommit_handler_maker.session_scope_key: str :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.sync_autocommit_handler_maker.session_scope_key: The key to use within the application state

Return type:

Callable[[Union[HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPServerPushEvent, HTTPDisconnectEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketResponseStartEvent, WebSocketResponseBodyEvent, WebSocketCloseEvent], Union[HTTPScope, WebSocketScope]], None]

Returns:

The handler callable

advanced_alchemy.extensions.litestar.sync_default_before_send_handler(message, scope)

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

Parameters:
Return type:

None

Returns:

None

advanced_alchemy.extensions.litestar.sync_default_handler_maker(session_scope_key='_sqlalchemy_db_session')

Set up the handler to issue a transaction commit or rollback based on specified status codes :type _sphinx_paramlinks_advanced_alchemy.extensions.litestar.sync_default_handler_maker.session_scope_key: str :param _sphinx_paramlinks_advanced_alchemy.extensions.litestar.sync_default_handler_maker.session_scope_key: The key to use within the application state

Return type:

Callable[[Union[HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPServerPushEvent, HTTPDisconnectEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketResponseStartEvent, WebSocketResponseBodyEvent, WebSocketCloseEvent], Union[HTTPScope, WebSocketScope]], None]

Returns:

The handler callable

Additional API References