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 portal: Portal: Get the portal.

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

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

init_app(app)[source]

Initialize the Flask application.

Parameters:

app¶ (Flask) – The Flask application to initialize.
app (Flask)

Raises:

ImproperConfigurationError – If the extension is already registered on the Flask application.

Return type:

None

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.
bind_key (str)

Returns:

A new session from the configured session factory.

Raises:

ImproperConfigurationError – If no session maker is found for the bind key.

Return type:

Union[AsyncSession, Session]

get_async_session(bind_key='default')[source]

Get an async session from the configured session factory.

Parameters:

bind_key¶ (str) – The bind key to use for the session.
bind_key (str)

Raises:

ImproperConfigurationError – If the session is not an async session.

Return type:

AsyncSession

Returns:

An async session from the configured session factory.

get_sync_session(bind_key='default')[source]

Get a sync session from the configured session factory.

Parameters:

bind_key¶ (str) – The bind key to use for the session.
bind_key (str)

Raises:

ImproperConfigurationError – If the session is not a sync session.

Return type:

Session

Returns:

A sync session from the configured session factory.

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.
bind_key (str)

Yields:

A session.

Return type:

Generator[Union[AsyncSession, Session], None, None]

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', toml_file=None, 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', toml_file=None, 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.

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

Parameters:: value (Any)
Return type:: str

json_deserializer(self, buf): For dialects that support the JSON datatype, this is a Python callable that will convert a JSON string to a Python object.

json_serializer()

For dialects that support the JSON datatype, this is a Python callable that will render a given object as JSON.

Return type:: str
Parameters:: value (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=<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'>)

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.
data (Any)
args (Any)
status_code (int)
kwargs (Any)

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.

app: Flask | None = None: The Flask application instance.

__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, bind_key=None, enable_touch_updated_timestamp_listener=True, enable_file_object_listener=True, file_object_raise_on_error=True, alembic_config=<factory>, routing_config=None, app=None, commit_mode='manual')

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

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.
app (Flask)
portal (Optional[Portal])

Raises:

ImproperConfigurationError – If portal is not provided for async configuration.

Return type:

None

close_engines(portal)[source]

Close the engines.

Parameters:

portal¶ (Portal) – The portal to use for thread-safe communication.
portal (Portal)

Return type:

None

async create_all_metadata()[source]

Create all metadata tables in the database.

Return type:: None

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.

app: Flask | None = None: The Flask application instance.

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

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.
app (Flask)
portal (Optional[Portal])

Return type:

None

close_engines(portal)[source]

Close the engines.

Parameters:

portal¶ (Portal) – The portal to use for thread-safe communication.
portal (Portal)

Return type:

None

create_all_metadata()[source]

Create all metadata tables in the database.

Return type:: None

__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, bind_key=None, enable_touch_updated_timestamp_listener=True, enable_file_object_listener=True, file_object_raise_on_error=True, alembic_config=<factory>, routing_config=None, app=None, commit_mode='manual')

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.
app (Flask)

Returns:

The Advanced Alchemy extension instance.

Return type:

AdvancedAlchemy

Raises:

advanced_alchemy.exceptions.ImproperConfigurationError – If the extension is not found.