flask¶

API Reference for the Flask extensions module

Note

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

Flask extension for Advanced Alchemy.

This module provides Flask integration for Advanced Alchemy, including session management, database migrations, and service utilities.

class advanced_alchemy.extensions.flask.AdvancedAlchemy[source]

Bases: object

Flask extension for Advanced Alchemy.

__init__(config, app=None, *, portal_provider=None)[source]: Initialize the extension.

property config: Sequence[SQLAlchemyAsyncConfig | SQLAlchemySyncConfig]: Get the SQLAlchemy configuration(s).

get_async_session(bind_key='default')[source]

Get an async session from the configured session factory.

Return type:: AsyncSession

get_session(bind_key='default')[source]

Get a new session from the configured session factory.

Parameters:: bind_key¶ – The bind key to use for the session.
Returns:: A new session from the configured session factory.
Raises:: ImproperConfigurationError – If no session maker is found for the bind key.

get_sync_session(bind_key='default')[source]

Get a sync session from the configured session factory.

Return type:: Session

init_app(app)[source]

Initialize the Flask application.

Parameters:: app¶ (Flask) – The Flask application to initialize.
Raises:: ImproperConfigurationError – If the extension is already registered on the Flask application.
Return type:: None

property is_async_enabled: bool: Return True if any of the database configs are async.

property portal: Portal: Get the portal.

with_session(bind_key='default')[source]

Provide a transactional scope around a series of operations.

Parameters:: bind_key¶ – The bind key to use for the session.
Yields:: A session.

class advanced_alchemy.extensions.flask.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.flask.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.flask.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.flask.EngineConfig[source]

Bases: EngineConfig

Configuration for SQLAlchemy’s Engine.

This class extends the base EngineConfig with Flask-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.

Calls the :func:schema_dump function to convert the value to a built-in before encoding.

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

class advanced_alchemy.extensions.flask.FlaskServiceMixin[source]

Bases: object

Flask service mixin.

This mixin provides Flask-specific functionality for services.

jsonify(data, *args, status_code=200, **kwargs)[source]

Convert data to a Flask JSON response.

Parameters:

data¶ (Any) – Data to serialize to JSON.
*args¶ (Any) – Additional positional arguments passed to Flask’s response class.
status_code¶ (int) – HTTP status code for the response. Defaults to 200.
**kwargs¶ (Any) – Additional keyword arguments passed to Flask’s response class.

Returns:

A Flask response with JSON content type.

Return type:

flask.Response

class advanced_alchemy.extensions.flask.SQLAlchemyAsyncConfig[source]

Bases: SQLAlchemyAsyncConfig

Flask-specific asynchronous SQLAlchemy configuration.

app: The Flask application instance.

commit_mode: The commit mode to use for database sessions.

__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')

app: Optional[Flask] = None: The Flask application instance.

close_engines(portal)[source]

Close the engines.

Parameters:: portal¶ (Portal) – The portal to use for thread-safe communication.
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[[], AsyncSession]

init_app(app, portal=None)[source]

Initialize the Flask application with this configuration.

Parameters:

app¶ – The Flask application instance.
portal¶ – The portal to use for thread-safe communication.

Raises:

ImproperConfigurationError – If portal is not provided for async configuration.

class advanced_alchemy.extensions.flask.SQLAlchemySyncConfig[source]

Bases: SQLAlchemySyncConfig

Flask-specific synchronous SQLAlchemy configuration.

app: The Flask application instance.

commit_mode: The commit mode to use for database sessions.

__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')

app: Optional[Flask] = None: The Flask application instance.

close_engines(portal)[source]

Close the engines.

Parameters:: portal¶ (Portal) – The portal to use for thread-safe communication.
Return type:: None

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

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]

init_app(app, portal=None)[source]

Initialize the Flask application with this configuration.

Parameters:

app¶ – The Flask application instance.
portal¶ – The portal to use for thread-safe communication. Unused in synchronous configurations.

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

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

Retrieve the Advanced Alchemy extension from the Flask application.

Parameters:: app¶ (Flask) – The flask.Flask application instance.
Returns:: The Advanced Alchemy extension instance.
Return type:: AdvancedAlchemy
Raises:: advanced_alchemy.exceptions.ImproperConfigurationError – If the extension is not found.