Flask-SQLAlchemy-Lite

https://flask-sqlalchemy-lite.readthedocs.io
Admin
Flask-SQLAlchemy-Lite is a Flask/Quart extension that manages SQLAlchemy engines and sessions for...
Tokens:9,409
Snippets:92
Trust Score:8.3
Update:10 months ago
Show doc for...
Tokens:
Raw
### Install Flask-SQLAlchemy-Lite using pip

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable

This command installs the Flask-SQLAlchemy-Lite package from PyPI, making it available for use in your Python project. It is a prerequisite for integrating the extension into your Flask or Quart application.

```bash
pip install Flask-SQLAlchemy-Lite

```

--------------------------------

### Install Flask-SQLAlchemy-Lite using pip

Source: https://flask-sqlalchemy-lite.readthedocs.io/index

This command installs the Flask-SQLAlchemy-Lite package from PyPI using pip, a standard package installer for Python. Ensure you have pip installed and configured in your environment.

```bash
pip install Flask-SQLAlchemy-Lite
```

--------------------------------

### Unittest Setup for Database Initialization

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/testing

This code snippet provides a setup function for Python's built-in 'unittest' framework. It initializes a test database by creating it if it doesn't exist and then populating the tables. It also registers a cleanup function to drop the database after tests are complete. Requires 'unittest', 'sqlalchemy_utils', and 'project' libraries.

```python
import unittest
from sqlalchemy_utils import create_database, drop_database, database_exists
from project import create_app, db, Model

def setUp():
    app = create_app({
        "SQLALCHEMY_ENGINES": {"default": "postgresql:///project-test"}
    })

    with app.app_context():
        engines = db.engines

    for engine in engines.values():
        if database_exists(engine.url):
            drop_database(engine.url)

        create_database(engine.url)

    Model.metadata.create_all(engines["default"])

    def drop_db():
        for engine in engines.values():
            drop_database(engine.url)

    unittest.addModuleCleanup(drop_db)

```

--------------------------------

### Setup Flask App with SQLAlchemy Instance (Direct Initialization)

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/start

Shows how to initialize the SQLAlchemy extension directly when creating the Flask app instance, bypassing the app factory pattern. It configures a default SQLite engine. Requires Flask and Flask-SQLAlchemy-Lite.

```python
app = Flask(__name__)
app.config |= {
    "SQLALCHEMY_ENGINES": {
        "default": "sqlite:///default.sqlite",
    },
}
app.config.from_prefixed_env()
db = SQLAlchemy(app)

```

--------------------------------

### Setup Flask App with SQLAlchemy Instance (App Factory)

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/start

Demonstrates setting up a Flask application using the app factory pattern and initializing the SQLAlchemy extension. It defines a default SQLite engine and configures the app. Requires the Flask and Flask-SQLAlchemy-Lite libraries.

```python
from flask import Flask
from flask_sqlalchemy_lite import SQLAlchemy

db = SQLAlchemy()

def create_app():
    app = Flask(__name__)
    app.config |= {
        "SQLALCHEMY_ENGINES": {
            "default": "sqlite:///default.sqlite",
        },
    }
    app.config.from_prefixed_env()
    db.init_app(app)
    return app

```

--------------------------------

### Flask-Alembic Setup with Flask-SQLAlchemy-Lite

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/alembic

Integrates Flask-Alembic with Flask and Flask-SQLAlchemy-Lite to manage database migrations. This setup requires defining SQLAlchemy models and initializing both SQLAlchemy and Alembic within the Flask application. It supports multiple database engines configured via Flask.

```python
from flask import Flask
from flask_alembic import Alembic
from flask_sqlalchemy_lite import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Model(DeclarativeBase):
    pass

class User(Model):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]

db = SQLAlchemy()
alembic = Alembic(metadatas=Model.metadata)

def create_app():
    app = Flask(__name__)
    app.config["SQLALCHEMY_ENGINES"] = {"default": "sqlite:///default.sqlite"}
    app.config.from_prefixed_env()
    db.init_app(app)
    alembic.init_app(app)
    return app

```

--------------------------------

### SQLite In-Memory Test Database Setup

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/testing

Configures a pytest fixture to use an in-memory SQLite database for testing. This approach is faster than file-based databases and allows for creating tables within the app fixture itself, simplifying test setup by combining app creation and database table initialization.

```python
import pytest
from project import create_app, db, Model

@pytest.fixture
def app():
    app = create_app({
        "SQLALCHEMY_ENGINES": {"default": "sqlite://"}
    })

    with app.app_context():
        engine = db.engine

    Model.metadata.create_all(engine)
    yield app

```

--------------------------------

### Get or Abort Entity (Sync)

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/api

Retrieves an entity by its primary key or aborts if not found. Supports options and custom arguments for session.get and abort. Defaults to a 404 error code.

```python
async_get_or_abort(_entity_ , _ident_ , _*_ , _options =None_, _get_kwargs =None_, _session =None_, _code =404_, _abort_kwargs =None_)
```

--------------------------------

### Create Database Tables within App Context

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/start

Shows how to create all defined database tables using metadata.create_all() within a Flask application context. This method is suitable for initial setup but does not handle table alterations. Requires a Flask app instance and SQLAlchemy.

```python
with app.app_context():
    Model.metadata.create_all(db.engine)
    OtherModel.metadata.create_all(db.get_engine("other"))

```

--------------------------------

### Get or Abort by Primary Key with Flask-SQLAlchemy

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/start

Illustrates using SQLAlchemy.get_or_abort() to retrieve a model instance by its primary key. If the instance does not exist, it automatically calls flask.abort() with a 404 error, commonly used for detail views.

```python
@app.get("/invoice/<int:id>")
def invoice_detail(id: int):
    obj: Invoice = db.get_or_abort(Invoice, id)
    ...

```

--------------------------------

### Manage Flask Application Context for DB Access

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/start

Explains how to access database engines and sessions within a Flask application context. It provides an example of pushing a context using a 'with' block for scenarios outside of requests or CLI commands.

```python
with app.app_context():
    # db.session and db.engine are accessible
    ...

```

--------------------------------

### Pytest Fixture for App Instance

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/testing

A pytest fixture that creates and yields a Flask application instance configured for testing. It utilizes the `create_app` factory with a specific test database URL, ensuring each test starts with a clean slate.

```python
import pytest
from project import create_app

@pytest.fixture
def app():
    app = create_app({
        "SQLALCHEMY_ENGINES": {"default": "sqlite://"}
    })
    yield app

```

--------------------------------

### Pytest Example: Testing Data Around Requests

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/testing

This pytest test function demonstrates how to insert data before a request, make a request to modify it, and then verify the changes within an app context. It highlights the need to manually push an app context for direct database operations before and after the request. Requires 'project' library.

```python
from project import db, User

def test_update_user(app):
    # Insert a user to be updated.
    with app.app_context():
        user = User(username="example", name="Example User")
        db.session.add(user)
        user_id = user.id

    # Make a request to the update endpoint. Outside the app context!
    client = app.test_client()
    client.post(f"/user/update/{user_id}", data={"name": "Real Name"})

    # Query the user and verify the update.
    with app.app_context():
        user = db.session.get(User, user_id)
        assert user.name == "Real Name"

```

--------------------------------

### Get or Abort by Natural Key with Flask-SQLAlchemy

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/start

Shows how to use SQLAlchemy.one_or_abort() to fetch a single row based on custom criteria defined in a select statement, instead of the primary key. It aborts if zero or more than one result is found.

```python
@app.get("/invoice/<company>/<date>")
def invoice_detail(company: str, date: str):
    query = select(Invoice).where(Invoice.company == company, Invoice.date == date)
    obj: Invoice = db.one_or_abort(query)
    ...

```

--------------------------------

### Pytest Test for Deactivating Old Users (with App Context Fixture)

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/testing

An example pytest test function that utilizes the 'app_ctx' fixture to directly interact with the database. This test verifies the functionality of a 'deactivate_old_users' method by inserting a user, committing, and then asserting the user's status after the method is called. It requires 'datetime', 'pytest', and 'project' libraries.

```python
from datetime import datetime, timedelta, UTC
import pytest
from project import db, User

@pytest.mark.usefixtures("app_ctx")
def test_deactivate_old_users():
    db.session.add(User(active=True, last_seen=datetime.now(UTC) - timedelta(days=32)))
    db.session.commit()
    # before running the deactivate job, there is one active user
    assert len(db.session.scalars(User).where(User.active).all()) == 1
    User.deactivate_old_users()  # a method you wrote
    # there are no longer any active users
    assert len(db.session.scalars(User).where(User.active).all()) == 0

```

--------------------------------

### Application Initialization

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/api

Registers the SQLAlchemy extension with a Flask application, creating engines from its configuration.

```APIDOC
## init_app Method

### Description
Registers the extension on an application, creating engines from its `config`.

### Method
`init_app(app)`

### Parameters
- **app** (App) - Required. The Flask application to register.

### Return Type
None
```

--------------------------------

### Plain Alembic Initialization

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/alembic

Initializes a new Alembic migration environment for a project. This command creates a `migrations` directory containing configuration files, including `env.py`, which needs to be modified for integration with the Flask application.

```bash
$ alembic init migrations

```

--------------------------------

### SQLAlchemy Initialization

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/api

Initializes the SQLAlchemy extension with a Flask application, setting up engines and sessions based on the application's configuration.

```APIDOC
## SQLAlchemy Class Initialization

### Description
Manages SQLAlchemy engines and sessions for Flask applications.

### Method
`__init__` (constructor)

### Parameters
- **app** (App | None) - Optional. The Flask application instance to initialize the extension with. If None, `init_app()` must be called later.
- **require_default_engine** (bool) - Optional. Defaults to True. If True, raises an error if a "default" engine is not configured.
- **engine_options** (dict[str, Any] | None) - Optional. Default arguments passed to `sqlalchemy.create_engine()` for each configured engine.
- **session_options** (dict[str, Any] | None) - Optional. Arguments to configure `sessionmaker` with.
```

--------------------------------

### Execute SQLAlchemy Queries with Flask-SQLAlchemy

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/start

Demonstrates how to add, modify, and query data using SQLAlchemy's select() constructor within Flask routes. It shows how to retrieve all users ordered by name and how to create a new user, ensuring uniqueness.

```python
from flask import request, abort, render_template
from sqlalchemy import select

@app.route("/users")
def user_list():
    users = db.session.scalars(select(User).order_by(User.name)).all()
    return render_template("users/list.html", users=users)

@app.route("/users/create")
def user_create():
    name = request.form["name"]

    if db.session.scalar(select(User).where(User.name == name)) is not None:
        abort(400)

    db.session.add(User(name=name))
    db.session.commit()
    return app.redirect(app.url_for("user_list"))

```

--------------------------------

### Plain Alembic CLI Commands

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/alembic

Command-line interface commands for Alembic to generate database revisions with autogenerate capabilities and apply them to the database.

```bash
$ alembic revision --autogenerate -m 'init'
$ alembic upgrade head

```

--------------------------------

### Configuring Multiple Database Binds with Session Options

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/session

Illustrates how to configure multiple database connections (binds) for different models or tables during the initialization of the SQLAlchemy extension. This allows for routing queries to specific databases based on the model. The `session_options` parameter accepts a dictionary mapping models to engine names.

```python
db = SQLAlchemy(session_options={"binds": {
    User: "auth",
    Role: "auth",
    ExternalBase: "external",
}})

```

--------------------------------

### Flask-Alembic CLI Commands

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/alembic

Command-line interface commands for Flask-Alembic to generate database revisions and apply them to the database.

```bash
$ flask db revision 'init'
$ flask db upgrade

```

--------------------------------

### Plain Alembic Configuration (`env.py`)

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/alembic

Modifies the `migrations/env.py` script to integrate with a Flask application and Flask-SQLAlchemy-Lite. It sets up the application context, target metadata, and database connection for Alembic to perform migrations.

```python
from project import create_app, Model, db

flask_app = create_app()

...

target_metadata = Model.metadata

...

def run_migrations_online() -> None:
    with flask_app.app_context():
        connectable = db.engine

    ...

...

```

--------------------------------

### Utilize Async SQLAlchemy Engines and Sessions in Flask

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/start

Details how to access asynchronous SQLAlchemy engines and sessions by prefixing with 'async_'. It highlights the need for careful session management in concurrent tasks, recommending specific strategies like get_async_session or async_sessionmaker.

```python
# Example: Accessing async session
async_session = db.async_session

# Example: Getting a named async session
named_async_session = db.get_async_session("my_unique_session_name")

# Example: Using async_sessionmaker (conceptual)
# async_session_factory = db.async_sessionmaker

```

--------------------------------

### Unittest TestCase for App Initialization and DB Isolation

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/testing

A basic unittest TestCase class that sets up the Flask application and enters a database isolation context for each test method. This ensures that tests using this TestCase have a clean database state and that changes are rolled back. Requires 'unittest' and 'project' libraries.

```python
import unittest

class AppTestCase(unittest.TestCase):
    def setUp(self):
        self.app = create_app({
            "SQLALCHEMY_ENGINES": {"default": "postgresql:///project-test"}
        })
        self.enterContext(db.test_isolation())

```

--------------------------------

### Manual Session Management with Context Manager

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/session

Demonstrates how to manually manage a session's lifecycle using a `with` block. This ensures the session is properly closed and any uncommitted state is handled automatically. It's crucial for sessions not tied to the request lifecycle.

```python
with db.sessionmaker() as session:
    ...
```

--------------------------------

### Configure Sync Engines with Flask Config

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/engine

Defines synchronous SQLAlchemy engine configurations using the `SQLALCHEMY_ENGINES` Flask config variable. Each key represents an engine name, and the value is either a connection string or a dictionary of arguments for `sqlalchemy.create_engine()`.

```python
SQLALCHEMY_ENGINES = {
    "default": "sqlite:///default.sqlite"
}
```

```python
from sqlalchemy import URL
SQLALCHEMY_ENGINES = {
    "default": URL.create("sqlite", database="default.sqlite")
}
```

```python
SQLALCHEMY_ENGINES = {
    "default": {"url": "sqlite:///default.sqlite"}
}
```

```python
from sqlalchemy import URL
SQLALCHEMY_ENGINES = {
    "default": {"url": URL.create("sqlite", database="default.sqlite")}
}
```

```python
SQLALCHEMY_ENGINES = {
    "default": {"url": {"drivername": "sqlite", "database": "default.sqlite"}}
}
```

--------------------------------

### Session Management

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/api

Facilitates the creation and management of SQLAlchemy sessions, including synchronous and asynchronous options, with automatic context management.

```APIDOC
## Session Properties and Methods

### Description
Provides access to session factories and managed sessions for the current application context.

### Properties
- **sessionmaker** (sessionmaker[Session]) - Read-only. The session factory configured for the current application. Can be used to create sessions directly or to update session options via its `configure()` method.
- **session** (Session) - Read-only. The default session for the current application context. It will be closed when the context ends.
- **async_sessionmaker** (async_sessionmaker[AsyncSession]) - Read-only. The asynchronous session factory configured for the current application. Preferred for direct management of async session lifetimes.
- **async_session** (AsyncSession) - Read-only. The default asynchronous session for the current application context. It will be closed when the context ends.

### Methods
- **get_session(name: str = 'default')** -> Session
  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.
  Parameters:
    - **name** (str) - A unique name for caching the session. Defaults to 'default'.
  Return type:
    Session
- **get_async_session(name: str = 'default')** -> AsyncSession
  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. Async sessions are not safe to use across concurrent tasks.
  Parameters:
    - **name** (str) - A unique name for caching the session. Defaults to 'default'.
  Return type:
    AsyncSession
```

--------------------------------

### Engine Management

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/api

Provides access to SQLAlchemy engines, both default and named, and their asynchronous counterparts.

```APIDOC
## Engine Properties and Methods

### Description
Provides access to the engines associated with the current application.

### Properties
- **engines** (dict[str, Engine]) - Read-only. The dictionary of all engines associated with the current application.
- **engine** (Engine) - Read-only. The default engine associated with the current application.
- **async_engines** (dict[str, AsyncEngine]) - Read-only. The dictionary of all asynchronous engines associated with the current application.
- **async_engine** (AsyncEngine) - Read-only. The default asynchronous engine associated with the current application.

### Methods
- **get_engine(name: str = 'default')** -> Engine
  Get a specific engine associated with the current application.
  Parameters:
    - **name** (str) - The name associated with the engine. Defaults to 'default'.
- **get_async_engine(name: str = 'default')** -> AsyncEngine
  Get a specific asynchronous engine associated with the current application.
  Parameters:
    - **name** (str) - The name associated with the engine. Defaults to 'default'.
```

--------------------------------

### Define SQLAlchemy Models with Type Annotations (SQLAlchemy 2)

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/start

Illustrates defining database models using SQLAlchemy 2's modern approach with type annotations. It includes a base model, a User model with a relationship to Post, and a Post model with a ForeignKey to User and a default timestamp. Requires SQLAlchemy and Flask-SQLAlchemy-Lite.

```python
from __future__ import annotations
from datetime import datetime
from datetime import UTC
from sqlalchemy import ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Model(DeclarativeBase):
    pass

class User(Model):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    posts: Mapped[list[Post]] = relationship(back_populates="author")

class Post(Model):
    __tablename__ = "post"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    body: Mapped[str]
    author_id: Mapped[int] = mapped_column(ForeignKey(User.id))
    author: Mapped[User] = relationship(back_populates="posts")
    created_at: Mapped[datetime] = mapped_column(default=lambda: datetime.now(UTC))

```

--------------------------------

### Interact with Database in Flask Shell

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/start

Demonstrates how to perform database operations within the Flask interactive shell. It shows querying users, modifying a field, and committing changes. The SQLAlchemy instance (`db`), the `User` model, and the `sqlalchemy` namespace (`sa`) are automatically available.

```python
>>> for user in db.session.scalars(sa.select(User)):
...     user.active = False
...
>>> db.session.commit()

```

--------------------------------

### Configure Async Engines with Flask Config

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/engine

Defines asynchronous SQLAlchemy engine configurations using the `SQLALCHEMY_ASYNC_ENGINES` Flask config variable. Similar to `SQLALCHEMY_ENGINES`, it accepts connection strings or dictionaries for `sqlalchemy.create_engine()` for async engines.

```python
SQLALCHEMY_ASYNC_ENGINES = {
    "async_default": "sqlite+aiosqlite:///async_default.sqlite"
}
```

--------------------------------

### Test Database Isolation (Async)

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/api

Asynchronous context manager for database test isolation. Isolates both sync and async operations, eliminating the need for 'test_isolation'. Commits are rolled back upon exiting the 'async with' block.

```python
async_test_isolation()
```

--------------------------------

### Test Database Isolation (Sync)

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/api

Provides a context manager for database test isolation. Commits are rolled back upon exiting the 'with' block, preventing interference with other tests. Patches SQLAlchemy engine and session to use a single connection within a transaction.

```python
test_isolation()
```

--------------------------------

### One or Abort Single Result (Sync)

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/api

Executes a select statement and returns a single result, or aborts if zero or multiple results are found. Can return a scalar model instance or a Row tuple. Defaults to a 404 error code.

```python
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_)
```

```python
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_)
```

--------------------------------

### Session Helper: get_or_abort

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/api

A utility method to retrieve an entity by its identifier from the session, aborting with a 404 error if not found.

```APIDOC
## get_or_abort Method

### Description
Calls `Session.get_one` and returns the instance. If not found, calls `abort()` with a 404 error by default.

### Method
`get_or_abort(entity, ident, *args, options=None, get_kwargs=None, session=None, code=404, abort_kwargs=None)`

### Parameters
- **entity** (type[M] | Mapper[M]) - Required. The model or mapper to query.
- **ident** - Required. The identifier of the entity to retrieve.
- ***args** - Positional arguments to pass to `Session.get_one`.
- **options** (dict) - Optional. Additional options for `Session.get_one`.
- **get_kwargs** (dict) - Optional. Keyword arguments for fetching the entity.
- **session** (Session) - Optional. The session to use. If None, the default session is used.
- **code** (int) - Optional. The HTTP status code to use for aborting if the entity is not found. Defaults to 404.
- **abort_kwargs** (dict) - Optional. Keyword arguments to pass to the `abort()` function.
```

--------------------------------

### Flask App Factory Pattern for Testing

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/testing

Defines a Flask application factory function that accepts test configuration overrides. This pattern is crucial for isolating database changes in tests by creating separate app instances. It initializes Flask-SQLAlchemy with the app.

```python
from flask import Flask
from flask_sqlalchemy_lite import SQLAlchemy

db = SQLAlchemy()

def create_app(test_config=None):
    app = Flask(__name__)
    app.config |= {
        "SQLALCHEMY_ENGINES": {"default": "sqlite:///default.sqlite"}
    }

    if test_config is None:
        app.config.from_prefixed_env()
    else:
        app.testing = True
        app.config |= test_config

    db.init_app(app)
    return app

```

--------------------------------

### Manage Test Database Lifecycle with Pytest Fixture

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/testing

A session-scoped, autouse pytest fixture to manage the lifecycle of a PostgreSQL test database. It drops existing databases, creates new ones, and initializes all model tables before tests run, then cleans up afterward. This ensures test isolation and prevents data corruption.

```python
import pytest
from sqlalchemy_utils import create_database, drop_database, database_exists
from project import create_app, db, Model

@pytest.fixture(scope="session", autouse=True)
def _manage_test_database():
    app = create_app({
        "SQLALCHEMY_ENGINES": {"default": "postgresql:///project-test"}
    })

    with app.app_context():
        engines = db.engines

    for engine in engines.values():
        if database_exists(engine.url):
            drop_database(engine.url)

        create_database(engine.url)

    Model.metadata.create_all(engines["default"])

    yield

    for engine in engines.values():
        drop_database(engine.url)

```

--------------------------------

### One or Abort Single Result (Async)

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/api

Asynchronous version of one_or_abort. Executes a select statement and returns a single result, or aborts if zero or multiple results are found. Can return a scalar model instance or a Row tuple. Defaults to a 404 error code.

```python
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_)
```

```python
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_)
```

--------------------------------

### Pytest Fixture for App Context

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/testing

This pytest fixture provides an application context for tests that need to interact with the database directly, without making HTTP requests. It ensures the app context is active for the duration of the test. It depends on a pre-existing 'app' fixture.

```python
import pytest

@pytest.fixture
def app_ctx(app):
    with app.app_context() as ctx:
        yield ctx

```

--------------------------------

### Pytest Fixture for Database Isolation

Source: https://flask-sqlalchemy-lite.readthedocs.io/en/stable/testing

This pytest fixture uses SQLAlchemy's test_isolation context manager to ensure that all database operations within a test are wrapped in a transaction, which is discarded after the test. This prevents tests from affecting each other's data. It requires the 'pytest' and 'project' libraries.

```python
import pytest
from project import create_app, db

@pytest.fixture
def app(monkeypatch):
    app = create_app({
        "SQLALCHEMY_ENGINES": {"default": "postgresql:///project-test"}
    })

    with db.test_isolation():
        yield app

```
Tokens:
Raw