sanic¶

API Reference for the Sanic extensions module

Note

Private methods and attributes are not included in the API reference.

class advanced_alchemy.extensions.sanic.AdvancedAlchemy[source]

Bases: Extension

Sanic extension for integrating Advanced Alchemy with SQLAlchemy.

Parameters:

config¶ – One or more configurations for SQLAlchemy.
app¶ – The Sanic application instance.

__init__(*, sqlalchemy_config, sanic_app=None)[source]

add_engine_dependency(engine_type, key=None)[source]

Add an engine dependency to the Sanic app.

Return type:: None

add_session_dependency(session_type, key=None)[source]

Add a session dependency to the Sanic app.

Return type:: None

get_async_config(key=None)[source]

Get the async config for the given key.

Return type:: SQLAlchemyAsyncConfig

get_async_engine(key=None)[source]

Get the async engine for the given key.

Return type:: sqlalchemy.ext.asyncio.AsyncEngine

get_async_session(request, key=None)[source]

Get the async session for the given key.

Return type:: AsyncSession

get_config(key=None)[source]

Get the config for the given key.

Return type:: Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig]

get_engine(key=None)[source]

Get the engine for the given key.

Return type:: Union[sqlalchemy.engine.Engine, sqlalchemy.ext.asyncio.AsyncEngine]

get_session(request, key=None)[source]

Get the session for the given key.

Return type:: Union[Session, AsyncSession]

get_sync_config(key=None)[source]

Get the sync config for the given key.

Return type:: SQLAlchemySyncConfig

get_sync_engine(key=None)[source]

Get the sync engine for the given key.

Return type:: sqlalchemy.engine.Engine

get_sync_session(request, key=None)[source]

Get the sync session for the given key.

Return type:: Session

map_configs()[source]

Maps the configs to the session bind keys.

Return type:: dict[str, Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig]]

provide_async_engine(key=None)[source]

Get the async engine for the given key.

Return type:: Callable[[], sqlalchemy.ext.asyncio.AsyncEngine]

provide_async_session(key=None)[source]

Get the async session for the given key.

Return type:: Callable[[Request], AsyncSession]

provide_engine(key=None)[source]

Get the engine for the given key.

Return type:: Callable[[], Union[sqlalchemy.engine.Engine, sqlalchemy.ext.asyncio.AsyncEngine]]

provide_session(key=None)[source]

Get the session for the given key.

Return type:: Callable[[Request], Union[Session, AsyncSession]]

provide_sync_engine(key=None)[source]

Get the sync engine for the given key.

Return type:: Callable[[], sqlalchemy.engine.Engine]

provide_sync_session(key=None)[source]

Get the sync session for the given key.

Return type:: Callable[[Request], Session]

register(sanic_app)[source]

Initialize the extension with the given Sanic app.

Return type:: None

property sanic_app: sanic.Sanic.(TypeAliasForwardRef('typing.Any'), TypeAliasForwardRef('typing.Any')): The Sanic app.

property sqlalchemy_config: Sequence[SQLAlchemyAsyncConfig | SQLAlchemySyncConfig]: Current Advanced Alchemy configuration.

startup(bootstrap)[source]

Advanced Alchemy Sanic extension startup hook.

Parameters:: bootstrap¶ (sanic_ext.Extend) – The Sanic extension bootstrap.
Return type:: None

with_async_session(key=None)[source]

Context manager for getting an async session.

Return type:: AsyncGenerator[AsyncSession, None]

with_sync_session(key=None)[source]

Context manager for getting a sync session.

Return type:: Generator[Session, None]

class advanced_alchemy.extensions.sanic.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.sanic.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.sanic.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'>)

sync_session_class: alias of Empty

class advanced_alchemy.extensions.sanic.EngineConfig[source]

Bases: EngineConfig

Configuration for SQLAlchemy’s Engine.

This class extends the base EngineConfig with Sanic-specific JSON serialization options.

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

json_deserializer: Callable for converting JSON strings to Python objects.

json_serializer: Callable for converting Python objects to JSON strings.

__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=<built-in method decode of msgspec.json.Decoder object>, 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_deserializer(self, buf)

Deserialize an object from JSON.

Parameters:: buf¶ (bytes-like or str) – The message to decode.
Returns:: obj – The deserialized object.
Return type:: Any

json_serializer()

Serialize JSON field values.

Parameters:: value¶ (Any) – Any JSON serializable value.
Returns:: JSON string representation of the value.
Return type:: str

class advanced_alchemy.extensions.sanic.SQLAlchemyAsyncConfig[source]

Bases: SQLAlchemyAsyncConfig

SQLAlchemy Async config for Sanic.

__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>, _app=None, commit_mode='manual', engine_key='db_engine', session_key='db_session', session_maker_key='session_maker_class')

property app: sanic.Sanic.(TypeAliasForwardRef('typing.Any'), TypeAliasForwardRef('typing.Any')): The Sanic application instance.

async close_engine()[source]

Close the engine.

Return type:: None

commit_mode: Literal['manual', 'autocommit', 'autocommit_include_redirect'] = 'manual': The commit mode to use for database sessions.

async create_all_metadata()[source]

Create all metadata tables in the database.

Return type:: None

create_session_maker()[source]

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

Returns:: Session factory used by the plugin.
Return type:: Callable[[], Session]

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

get_engine_from_request(request)[source]

Retrieve the engine from the request context.

Parameters:: request¶ (sanic.Request) – The incoming request.
Returns:: The SQLAlchemy engine.
Return type:: AsyncEngine

get_session_from_request(request)[source]

Retrieve the session from the request context.

Parameters:: request¶ (sanic.Request) – The incoming request.
Returns:: The session associated with the request.
Return type:: SessionT

get_sessionmaker_from_request(request)[source]

Retrieve the session maker from the request context.

Parameters:: request¶ (sanic.Request) – The incoming request.
Returns:: The session maker.
Return type:: SessionMakerT

init_app(app, bootstrap)[source]

Initialize the Sanic application with this configuration.

Parameters:

app¶ (Any'))) – The Sanic application instance.
bootstrap¶ (Extend) – The Sanic extension bootstrap.

Return type:

None

async on_shutdown()[source]

Handles the shutdown event by disposing of the SQLAlchemy engine.

Ensures that all connections are properly closed during application shutdown.

Return type:: None
Returns:: None

async on_startup()[source]

Initialize the Sanic application with this configuration.

Return type:: None

async session_handler(session, request, response)[source]

Handles the session after a request is processed.

Applies the commit strategy and ensures the session is closed.

Parameters:

session¶ (sqlalchemy.ext.asyncio.AsyncSession) – The database session.
request¶ (sanic.Request) – The incoming HTTP request.
response¶ (sanic.HTTPResponse) – The outgoing HTTP response.

Return type:

None

Returns:

None

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

session_maker_key: str = 'session_maker_class': Key under which to store the SQLAlchemy sessionmaker in the application state instance.

startup(bootstrap)[source]

Initialize the Sanic application with this configuration.

Parameters:: bootstrap¶ (Extend) – The Sanic extension bootstrap.
Return type:: None

engine_config: EngineConfig

Configuration for the SQLAlchemy engine.

The configuration options are documented in the SQLAlchemy documentation.

class advanced_alchemy.extensions.sanic.SQLAlchemySyncConfig[source]

Bases: SQLAlchemySyncConfig

SQLAlchemy Sync config for Starlette.

__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>, _app=None, commit_mode='manual', engine_key='db_engine', session_key='db_session', session_maker_key='session_maker_class')

property app: sanic.Sanic.(TypeAliasForwardRef('typing.Any'), TypeAliasForwardRef('typing.Any')): The Sanic application instance.

async close_engine()[source]

Close the engine.

Return type:: None

commit_mode: Literal['manual', 'autocommit', 'autocommit_include_redirect'] = 'manual': The commit mode to use for database sessions.

async create_all_metadata()[source]

Create all metadata tables in the database.

Return type:: None

create_session_maker()[source]

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

Returns:: Session factory used by the plugin.
Return type:: Callable[[], Session]

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

get_engine_from_request(request)[source]

Retrieve the engine from the request context.

Parameters:: request¶ (sanic.Request) – The incoming request.
Returns:: The SQLAlchemy engine.
Return type:: AsyncEngine

get_session_from_request(request)[source]

Retrieve the session from the request context.

Parameters:: request¶ (sanic.Request) – The incoming request.
Returns:: The session associated with the request.
Return type:: SessionT

get_sessionmaker_from_request(request)[source]

Retrieve the session maker from the request context.

Parameters:: request¶ (sanic.Request) – The incoming request.
Returns:: The session maker.
Return type:: SessionMakerT

init_app(app, bootstrap)[source]

Initialize the Sanic application with this configuration.

Parameters:

app¶ (Any'))) – The Sanic application instance.
bootstrap¶ (Extend) – The Sanic extension bootstrap.

Return type:

None

async on_shutdown()[source]

Handles the shutdown event by disposing of the SQLAlchemy engine.

Ensures that all connections are properly closed during application shutdown.

Return type:: None
Returns:: None

async on_startup()[source]

Initialize the Sanic application with this configuration.

Return type:: None

async session_handler(session, request, response)[source]

Handles the session after a request is processed.

Applies the commit strategy and ensures the session is closed.

Parameters:

session¶ (sqlalchemy.orm.Session) – The database session.
request¶ (sanic.Request) – The incoming HTTP request.
response¶ (sanic.HTTPResponse) – The outgoing HTTP response.

Return type:

None

Returns:

None

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

session_maker_key: str = 'session_maker_class': Key under which to store the SQLAlchemy sessionmaker in the application state instance.

startup(bootstrap)[source]

Initialize the Sanic application with this configuration.

Parameters:: bootstrap¶ (Extend) – The Sanic extension bootstrap.
Return type:: None

engine_config: EngineConfig

Configuration for the SQLAlchemy engine.

The configuration options are documented in the SQLAlchemy documentation.

class advanced_alchemy.extensions.sanic.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'>)