API

Anything documented here is part of the public API that Flask-SQLAlchemy provides, unless otherwise indicated. Anything not documented here is considered internal or private and may change at any time.

class flask_sqlalchemy_lite.SQLAlchemy(app=None, *, require_default_engine=True, engine_options=None, session_options=None)

Manage SQLAlchemy engines and sessions for Flask applications.

Parameters:

  • app (App | None) – Call init_app() on this Flask application.

  • require_default_engine (bool) – Whether to raise an error if a "default" engine is not configured.

  • engine_options (dict[str, t.Any] | None) – Default arguments passed to sqlalchemy.create_engine() for each configured engine.

  • session_options (dict[str, t.Any] | None) – Arguments to configure sessionmaker with.

init_app(app)

Register the extension on an application, creating engines from its config.

Parameters:

app (App) – The application to register.

Return type:

None

property engines: dict[str, Engine]

The engines associated with the current application.

get_engine(name='default')

Get a specific engine associated with the current application.

The engine attribute is a shortcut for calling this without an argument to get the default engine.

Parameters:

name (str) – The name associated with the engine.

Return type:

Engine

property engine: Engine

The default engine associated with the current application.

property sessionmaker: sessionmaker[Session]

The session factory configured for the current application. This can be used to create sessions directly, but they will not be closed automatically at the end of the application context. Use session and get_session() for that.

This can also be used to update the session options after init_app(), by calling its configure() method.

get_session(name='default')

Create a sqlalchemy.orm.Session that will be closed at the end of the application context. Repeated calls with the same name within the same application context will return the same session.

The session attribute is a shortcut for calling this without an argument to get the default session.

Parameters:

name (str) – A unique name for caching the session.

Return type:

Session

property session: Session

The default session for the current application context. It will be closed when the context ends.

property async_engines: dict[str, AsyncEngine]

The async engines associated with the current application.

get_async_engine(name='default')

Get a specific async engine associated with the current application.

The async_engine attribute is a shortcut for calling this without an argument to get the default engine.

Parameters:

name (str) – The name associated with the engine.

Return type:

AsyncEngine

property async_engine: AsyncEngine

The default async engine associated with the current application.

property async_sessionmaker: async_sessionmaker[AsyncSession]

The async session factory configured for the current application. This can be used to create sessions directly, but they will not be closed automatically at the end of the application context. This is the preferred way to use async sessions, by directly creating and managing them.

async_session and get_async_session() are available to manage async sessions for the entire application context, but managing them directly with this is preferred.

This can also be used to update the session options after init_app(), by calling its configure() method.

get_async_session(name='default')

Create a sqlalchemy.ext.asyncio.AsyncSession that will be closed at the end of the application context. Repeated calls with the same name within the same application context will return the same session.

The async_session attribute is a shortcut for calling this without an argument to get the default async session.

Async sessions are not safe to use across concurrent tasks, as they represent unguarded mutable state. Use a separate named session for each task within a single application context, or use async_sessionmaker to directly control async session lifetime.

Parameters:

name (str) – A unique name for caching the session.

Return type:

AsyncSession

property async_session: AsyncSession

The default async session for the current application context. It will be closed when the context ends.

get_or_abort(entity, ident, *, options=None, get_kwargs=None, session=None, code=404, abort_kwargs=None)

Call Session.get_one and return the instance. If not found, call abort() with a 404 error by default.

Parameters:

  • entity (type[M] | Mapper[M]) – The model to query.

  • ident (Any) – The primary key to query.

  • options (Sequence[ORMOption] | None) – The options argument passed to session.get.

  • get_kwargs (dict[str, Any] | None) – Arguments passed to session.get.

  • session (Session | None) – The session to execute the query. Defaults to session.

  • code (int) – The HTTP error code.

  • abort_kwargs (dict[str, Any] | None) – Arguments passed to abort.

Return type:

M

Added in version 0.2.

async async_get_or_abort(entity, ident, *, options=None, get_kwargs=None, session=None, code=404, abort_kwargs=None)

Async version of get_or_abort().

Added in version 0.2.

Parameters:
Return type:

M

one_or_abort(select: Select, *, scalar: Literal[True] = True, execute_kwargs: dict[str, Any] | None = None, session: Session | None = None, code: int = 404, abort_kwargs: dict[str, Any] | None = None) → M
one_or_abort(select: Select, *, scalar: Literal[False], execute_kwargs: dict[str, Any] | None = None, session: Session | None = None, code: int = 404, abort_kwargs: dict[str, Any] | None = None) → Row[TP]

Call Session.execute with the given select statement and return a single result. If zero or multiple results are found, call abort() with a 404 error by default.

This is useful instead of get_or_abort() if you are querying by some other unique key rather than the primary key.

execute will return tuples, even if you are selecting a single model class. By default, this function assumes you are querying a single model and calls scalar_one() to return the single instance.

Parameters:

  • select (Select) – The select statement to execute.

  • scalar (bool) – Whether to call scalar_one() on the result to return a scalar instead of a tuple.

  • execute_kwargs (dict[str, Any] | None) – Arguments passed to session.execute.

  • session (Session | None) – The session to execute the query. Defaults to session.

  • code (int) – The HTTP error code.

  • abort_kwargs (dict[str, Any] | None) – Arguments passed to abort.

Return type:

M | Row[TP]

Added in version 0.2.

async async_one_or_abort(select: Select, *, scalar: Literal[True] = True, execute_kwargs: dict[str, Any] | None = None, session: AsyncSession | None = None, code: int = 404, abort_kwargs: dict[str, Any] | None = None) → M
async async_one_or_abort(select: Select, *, scalar: Literal[False], execute_kwargs: dict[str, Any] | None = None, session: AsyncSession | None = None, code: int = 404, abort_kwargs: dict[str, Any] | None = None) → Row[TP]

Async version of one_or_abort().

Added in version 0.2.

Parameters:
Return type:

M | Row[TP]

test_isolation()

Context manager to isolate the database during a test. Commits are rolled back when the with block exits, and will not be seen by other tests.

This patches the SQLAlchemy engine and session to use a single connection, in a transaction that is rolled back when the context exits.

If the code being tested uses async features, use async_test_isolation() instead. It will isolate both the sync and async operations.

When using SQLite, follow the SQLAlchemy docs to fix the driver’s transaction handling.

Added in version 0.2.

Return type:

Iterator[None]

async_test_isolation()

Async version of test_isolation() to be used as an async with block. It will isolate the sync code as well, so you do not need to use test_isolation as well.

Added in version 0.2.

Return type:

AsyncIterator[None]