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.
- __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'>)¶
- 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=<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_session(state, scope)[source]¶
Create a session instance.
- Parameters:
- Return type:
- 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.
- 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)[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:
- classmethod detect_nested_field(field_definition)[source]¶
Return
True
iffield_definition
represents a nested model field.- Parameters:
field_definition¶ (
FieldDefinition
) – inspect type to determine if field represents a nested model.- Return type:
- 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
.
- 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 rootclick.Group
of the Litestar CLI- Return type:
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)[source]¶
Initialize
SQLAlchemyPlugin
.- Parameters:
config¶ (
SQLAlchemyAsyncConfig
|SQLAlchemySyncConfig
|list
[SQLAlchemyAsyncConfig
|SQLAlchemySyncConfig
]) – configure DB connection and hook handlers and dependencies.
- class advanced_alchemy.extensions.litestar.SQLAlchemySerializationPlugin[source]¶
Bases:
SerializationPluginProtocol
,SlotsBase
- 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:
- 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 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=<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)¶
- 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.
- 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:
message¶ (
Union
[HTTPResponseStartEvent
,HTTPResponseBodyEvent
,HTTPServerPushEvent
,HTTPDisconnectEvent
,WebSocketAcceptEvent
,WebSocketSendEvent
,WebSocketResponseStartEvent
,WebSocketResponseBodyEvent
,WebSocketCloseEvent
]) – ASGI-litestar.types.Message
scope¶ (
Union
[HTTPScope
,WebSocketScope
]) – An ASGI-litestar.types.Scope
- Return type:
- 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:
message¶ (
Union
[HTTPResponseStartEvent
,HTTPResponseBodyEvent
,HTTPServerPushEvent
,HTTPDisconnectEvent
,WebSocketAcceptEvent
,WebSocketSendEvent
,WebSocketResponseStartEvent
,WebSocketResponseBodyEvent
,WebSocketCloseEvent
]) – ASGI-Message
scope¶ (
Union
[HTTPScope
,WebSocketScope
]) – An ASGI-Scope
- Return type:
- 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:
- advanced_alchemy.extensions.litestar.sync_autocommit_before_send_handler(message, scope)¶
Handle commit/rollback, closing and cleaning up sessions before sending.
- Parameters:
message¶ (
Union
[HTTPResponseStartEvent
,HTTPResponseBodyEvent
,HTTPServerPushEvent
,HTTPDisconnectEvent
,WebSocketAcceptEvent
,WebSocketSendEvent
,WebSocketResponseStartEvent
,WebSocketResponseBodyEvent
,WebSocketCloseEvent
]) – ASGI-Message
scope¶ (
Union
[HTTPScope
,WebSocketScope
]) – An ASGI-Scope
- Return type:
- 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:
message¶ (
Union
[HTTPResponseStartEvent
,HTTPResponseBodyEvent
,HTTPServerPushEvent
,HTTPDisconnectEvent
,WebSocketAcceptEvent
,WebSocketSendEvent
,WebSocketResponseStartEvent
,WebSocketResponseBodyEvent
,WebSocketCloseEvent
]) – ASGI-Message
scope¶ (
Union
[HTTPScope
,WebSocketScope
]) – An ASGI-Scope
- Return type:
- 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