vs-db

Async database layer for Python — built on SQLAlchemy async. Designed to feel like Spring JPA: define your entity, extend a repo, annotate with @transactional, and never manually manage a session again.

---

The Problem It Solves

Without vs-db, every async database operation requires you to manually manage sessions, transactions, commits, and rollbacks:

# Without vs-db — repeated in every service method
async def create_user(name: str, email: str):
    async with async_session() as session:
        async with session.begin():
            user = User(name=name, email=email)
            session.add(user)
            await session.flush()
            await session.refresh(user)
            return user

With vs-db:

# With vs-db — session managed automatically
@transactional
async def create_user(self, name: str, email: str) -> User:
    return await self.create(User(name=name, email=email))

The middleware opens the session at the start of each request, commits on success, rolls back on exception, and closes it — all without you writing a single line of session management code.

---

Installation

pip install vs-db

Install with your database driver:

pip install vs-db[mysql]      # MySQL / MariaDB
pip install vs-db[postgres]   # PostgreSQL
pip install vs-db[sqlite]     # SQLite

Install the Starlette middleware support:

pip install vs-db[starlette]

---

How It All Fits Together

Application Startup
    └── VsDbSessionFactory(config)       # create engine + session factory once

Per Request (via middleware)
    └── VsDbMiddleware
            ├── creates AsyncSession
            ├── stores in ContextVar     # available everywhere in the request
            ├── commits on success
            ├── rolls back on exception
            └── closes session

Your Code
    └── @transactional on service/repo methods
            ├── pulls session from ContextVar (REQUIRED)
            └── or opens its own (REQUIRES_NEW / background tasks)

---

Application Startup

Initialise once before handling any request. Pass entities to tell the factory which packages contain your @entity classes — it scans and imports them automatically so they are registered before any table creation or query:

from vs_common.config.vs_ini_config import VsIniConfig
from vs_db.session.vs_db_session_factory import VsDbSessionFactory

config = VsIniConfig("config.ini")
db = VsDbSessionFactory(config, entities=["my_service.models", "my_service.audit"])

This mirrors Spring's @EntityScan — entity classes are discovered at startup regardless of whether you use create_tables() or an external migration tool like Alembic.

Auto Table Creation & Schema Validation (v0.1.1)

Call create_tables() at startup to automatically create all tables registered via @entity. If a table already exists, it is skipped — but the schema is validated against the entity definition and any drift is logged as a warning.

await db.create_tables()

Example warnings when entity and DB table are out of sync:

WARNING [User] Table 'users' is missing columns: {'first_name', 'last_name'}
WARNING [User] Table 'users' has extra columns not in entity: {'full_name'}

No columns are added or removed automatically — create_tables() only creates missing tables. Use a migration tool (e.g. Alembic) for schema changes on existing tables.

---

Data Versioning — @db_install & populate_data()

For seeding and managing initial data across application versions. This is separate from schema migrations (Alembic) — it handles the data that must exist before your application starts.

The library maintains a vs_db_version table to track the applied data version. On startup, populate_data() compares the version in your config against the DB and runs the required upgrade or downgrade steps in order.

Step 1 — Define version classes

Each class handles one version's data changes. The version string defines the execution order — any depth works (1.0, 1.0.0, 2.1.3).

# my_app/versions/v1_0_0.py
from vs_db.version.vs_base_version import VsBaseVersion
from vs_db.version.vs_db_install import db_install
from vs_db.decorator.vs_db_decorator import transactional

@db_install(version="1.0.0")
class V1_0_0(VsBaseVersion):

    @transactional
    async def upgrade(self) -> None:
        await self.db.execute(text(
            "INSERT INTO roles (name) VALUES ('admin'), ('user')"
        ))

    @transactional
    async def downgrade(self) -> None:
        await self.db.execute(text("DELETE FROM roles"))


@db_install(version="1.1.0")
class V1_1_0(VsBaseVersion):

    @transactional
    async def upgrade(self) -> None:
        await self.db.execute(text(
            "INSERT INTO settings (key, value) VALUES ('theme', 'light')"
        ))

    @transactional
    async def downgrade(self) -> None:
        await self.db.execute(text("DELETE FROM settings WHERE key = 'theme'"))

Step 2 — Register packages and set target version in config

VsDbSessionFactory(
    config,
    entities=["my_app.entity"],
    db_upgrade=["my_app.versions"],   # scanned for @db_install classes
)

[database]
url     = mysql+aiomysql://user:pass@localhost/mydb
version = 1.1.0

Step 3 — Call populate_data() at startup after create_tables()

await db.create_tables()
await db.populate_data()

How it works:

• populate_data() reads database.version from config (the target).
• Reads the latest entry from vs_db_version table (the current DB state).
• If DB version < config version → runs upgrade() for each missing version in ascending order.
• If DB version > config version → runs downgrade() for each excess version in descending order.
• If versions match → no-op.
• Each step runs in its own transaction. If a step fails, it stops and the successfully completed steps remain committed.
• After each step, the vs_db_version table is updated with the new current version.

Upgrade sequence example:

DB is at 1.0.0, config says 1.2.0:

Running upgrade → 1.1.0 ... ✓
Running upgrade → 1.2.0 ... ✓

Downgrade sequence example:

DB is at 1.2.0, config says 1.0.0:

Running downgrade → 1.2.0 ... (now at 1.1.0) ✓
Running downgrade → 1.1.0 ... (now at 1.0.0) ✓

config.ini:

[database]
url          = mysql+aiomysql://user:pass@localhost/mydb
pool_size    = 10
max_overflow = 20
echo         = false

Key	Default	Description
database.url	—	Required. Database connection URL including driver
database.pool_size	10	Number of persistent connections kept open
database.max_overflow	20	Extra connections allowed when pool is exhausted
database.echo	false	Log all SQL statements (dev only)
database.version	—	Target data version for populate_data(). If not set, populate_data() is a no-op

Max concurrent connections = pool_size + max_overflow. Connections are pre-pinged before use (pool_pre_ping=True) so stale connections are replaced transparently.

Example URLs for other databases:

# PostgreSQL
url = postgresql+asyncpg://user:pass@localhost/mydb

# SQLite
url = sqlite+aiosqlite:///./app.db

---

Middleware

Register once with your Starlette / FastAPI app:

from vs_db.middleware.vs_db_middleware import VsDbMiddleware

app.add_middleware(VsDbMiddleware)

What it does per request automatically:

Step	Action
Before handler	Creates AsyncSession, stores in ContextVar
After handler (success)	Commits transaction
After handler (exception)	Rolls back transaction
After commit	Flushes any deferred @cache_evict patterns
Finally	Closes session and clears ContextVar

You write zero session management code anywhere in your services or repos.

---

Defining Entities

from sqlalchemy.orm import Mapped, mapped_column
from vs_db.base.vs_db_base import VsDbBase
from vs_db.decorator.vs_db_decorator import entity

@entity("users")
class User(VsDbBase):
    id:    Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name:  Mapped[str]
    email: Mapped[str]

@entity("table_name") replaces the boilerplate __tablename__ class attribute and registers the model with SQLAlchemy's metadata automatically.

---

Mixins

Mixins are fully opt-in. Add only what you need:

from vs_db.base.vs_audit_mixin import AuditMixin
from vs_db.base.vs_soft_delete_mixin import SoftDeleteMixin
from vs_db.base.vs_version_mixin import VersionMixin

@entity("users")
class User(VsDbBase, AuditMixin, SoftDeleteMixin, VersionMixin):
    id:    Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name:  Mapped[str]
    email: Mapped[str]

AuditMixin

Adds created_at and updated_at. Both are set automatically via SQLAlchemy event hooks — you never touch them.

user.created_at   # set on INSERT
user.updated_at   # updated on every UPDATE

Without AuditMixin you would add these columns to every model and wire up event listeners or override save() in every repo. With the mixin: just inherit it.

SoftDeleteMixin

Adds a deleted_at column. Call repo.soft_delete(id) instead of repo.delete(id) to mark a record deleted without removing it from the database.

user.is_deleted   # True if deleted_at is set, False otherwise

Useful for audit trails, recovery scenarios, and compliance requirements where hard deletes are not acceptable.

VersionMixin

Adds a version column for optimistic locking. SQLAlchemy auto-increments version on every update. If two concurrent transactions read the same record and both try to update it, the second one raises StaleDataError — preventing silent data overwrites.

# Concurrent update scenario:
# Transaction A reads user (version=1), Transaction B reads user (version=1)
# Transaction A updates → version becomes 2, commits
# Transaction B tries to update → raises StaleDataError (version mismatch)

Without VersionMixin you would write custom version-check logic in every update path. With the mixin: SQLAlchemy handles it entirely.

---

Repository

Extend VsDbRepo[T] for your entity. You get all CRUD and query methods for free:

from vs_db.base.vs_db_repo import VsDbRepo
from vs_db.decorator.vs_db_decorator import transactional

class UserRepo(VsDbRepo[User]):

    def __init__(self):
        super().__init__(User)

    @transactional
    async def find_by_email(self, email: str) -> User | None:
        return await self.find_by_field(User.email, email)

    @transactional
    async def find_active(self) -> list[User]:
        return await self.find_all_by_field(User.is_deleted, False)

Built-in Methods

Method	Description
create(entity)	Insert and return with generated ID
insert_all(entities)	Bulk insert
get_by_id(id)	Fetch by primary key
get_all(skip, limit)	Offset-based list
get_page(page, page_size)	Returns Page[T] with total and page metadata
update(entity)	Merge and flush
delete(id)	Hard delete by primary key
soft_delete(id)	Sets deleted_at — requires SoftDeleteMixin
exists(id)	Returns True / False
count()	Total row count
find_by_field(field, value)	First result matching field
find_all_by_field(field, value)	All results matching field
upsert(entity, conflict_fields)	Insert or update on conflict — MySQL / MariaDB

Pagination

page = await repo.get_page(page=2, page_size=20)

page.items      # List[User] for page 2
page.total      # total rows across all pages
page.pages      # total number of pages
page.page       # current page number
page.page_size  # page size used

---

@transactional

The core decorator. Annotate any async method on a service or repo — session management happens automatically.

from vs_db.decorator.vs_db_decorator import transactional

@transactional
async def save_order(self, order: Order) -> Order:
    return await self.repo.create(order)

How it works:

• If a session is already in ContextVar (i.e. a request is active), it reuses it and wraps the call in a savepoint — a nested transaction that rolls back independently if the method raises.
• If no session exists (background task, script, test), it opens a standalone session, commits on success, and rolls back on exception.

This mirrors Spring's @Transactional(propagation = REQUIRED) behaviour exactly.

Propagation

from vs_db.decorator.vs_db_decorator import Propagation

@transactional(propagation=Propagation.REQUIRES_NEW)
async def write_audit_log(self, action: str) -> None:
    await self.audit_repo.create(AuditLog(action=action))

Value	Behaviour
REQUIRED (default)	Reuses existing session; opens new one if none exists
REQUIRES_NEW	Always opens a new independent session; suspends the current one

Use REQUIRES_NEW when you need a method to commit independently — e.g. audit logging that must persist even if the outer transaction rolls back.

Isolation

from vs_db.decorator.vs_db_decorator import Isolation

@transactional(isolation=Isolation.REPEATABLE_READ)
async def transfer(self, from_id: int, to_id: int, amount: float) -> None:
    ...

@transactional(isolation=Isolation.SERIALIZABLE)
async def allocate_seat(self, flight_id: int, seat: str) -> bool:
    ...

Value	Behaviour
DEFAULT	Database default (typically READ COMMITTED)
READ_COMMITTED	Sees only committed rows — prevents dirty reads
REPEATABLE_READ	Same snapshot throughout the transaction — prevents non-repeatable reads
SERIALIZABLE	Fully isolated — prevents phantom reads; use for financial or inventory operations

Custom Rollback Targets

By default all exceptions trigger a rollback. Narrow it down:

@transactional(rollback_on=(ValueError, RuntimeError))
async def process(self, data: dict) -> None:
    ...

---

Without @transactional — Manual Session Usage

You do not have to use @transactional at all. If you prefer explicit control, access the session directly from ContextVar and manage the transaction yourself.

Direct session access

When the middleware is active (inside a request), the session is always available:

from vs_db.context.db_context import get_session
from sqlalchemy import select

async def custom_query() -> list[User]:
    session = get_session()
    result = await session.execute(
        select(User).where(User.status == "active").order_by(User.created_at.desc())
    )
    return list(result.scalars().all())

Use this when you need a raw SQLAlchemy query that does not fit the base repo methods.

Manual transaction control via VsTransactionManager

from vs_db.transaction.vs_transaction_manager import VsTransactionManager

session = VsTransactionManager.get_session()   # session from ContextVar
VsTransactionManager.is_active()               # True if session is in context
await VsTransactionManager.commit()            # commit current transaction
await VsTransactionManager.rollback()          # rollback current transaction

Full manual example — equivalent to @transactional but written explicitly:

from vs_db.transaction.vs_transaction_manager import VsTransactionManager
from vs_db.context.db_context import get_session

async def transfer(from_id: int, to_id: int, amount: float) -> None:
    session = get_session()
    try:
        sender = await session.get(Account, from_id)
        receiver = await session.get(Account, to_id)
        sender.balance -= amount
        receiver.balance += amount
        await session.flush()
        await VsTransactionManager.commit()
    except Exception:
        await VsTransactionManager.rollback()
        raise

Mixing manual and annotated code

You can mix both freely. Methods with @transactional reuse whatever session is already in ContextVar — so calling an annotated repo method from manually managed code works without any changes:

async def process(user_id: int) -> None:
    session = get_session()                       # get session directly
    user = await session.get(User, user_id)       # raw query
    user.status = "processing"
    await session.flush()

    await order_repo.cancel_pending(user_id)      # @transactional method — reuses same session

    await VsTransactionManager.commit()           # one commit covers everything

---

Background Tasks and Scripts

Outside a request (scheduled jobs, CLI scripts, workers), there is no middleware to open a session. Use an unmanaged session:

from vs_db.transaction.vs_transaction_manager import VsTransactionManager

async def run_nightly_sync():
    await VsTransactionManager.open_unmanaged_session()
    try:
        repo = UserRepo()
        users = await repo.get_all()
        # process users ...
        await VsTransactionManager.commit()
    except Exception:
        await VsTransactionManager.rollback()
        raise
    finally:
        await VsTransactionManager.close_unmanaged_session()

@transactional methods called inside this block reuse the session exactly as they would in a request.

---

@model — API Schema from Entity

Pydantic request and response models must be defined separately from SQLAlchemy entities. The @model decorator eliminates the boilerplate by injecting fields from an entity into a Pydantic class automatically.

Fields that are nullable=True on the entity are injected as Optional[T] = None. Fields that are nullable=False are injected as required. model_config = {"from_attributes": True} is added automatically.

from pydantic import BaseModel, Field
from vs_db.decorator.vs_model_decorator import model

@model(AppUser, exclude=["password", "deleted_at"])
class UserResponse(BaseModel):
    pass
# AppUser fields injected: id (str), email (str), first_name (str), last_name (Optional[str])

@model(AppUser, include=["email", "first_name", "last_name"])
class CreateUserRequest(BaseModel):
    first_name: str = Field(min_length=1, max_length=50)  # user-defined — decorator skips it

Rules:

• If include is provided, only those fields are injected. If exclude is provided, all fields except those are injected. include takes priority if both are provided.
• Fields already defined on the class are never overridden — user definitions always win.
• Only mapped columns are considered. Relationships are ignored.

@model parameters:

Parameter	Type	Description
entity	SQLAlchemy entity class	The @entity-decorated class to derive fields from
include	List[str]	Whitelist of field names to inject. If set, only these fields are injected.
exclude	List[str]	Blacklist of field names to skip. Ignored if include is set.

---

Full Example

from sqlalchemy.orm import Mapped, mapped_column
from vs_db.base.vs_db_base import VsDbBase
from vs_db.base.vs_audit_mixin import AuditMixin
from vs_db.base.vs_soft_delete_mixin import SoftDeleteMixin
from vs_db.base.vs_db_repo import VsDbRepo
from vs_db.decorator.vs_db_decorator import entity, transactional, Propagation


@entity("orders")
class Order(VsDbBase, AuditMixin, SoftDeleteMixin):
    id:         Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    user_id:    Mapped[int]
    total:      Mapped[float]
    status:     Mapped[str]


@entity("audit_logs")
class AuditLog(VsDbBase, AuditMixin):
    id:      Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    action:  Mapped[str]
    user_id: Mapped[int]


class OrderRepo(VsDbRepo[Order]):
    def __init__(self):
        super().__init__(Order)

    @transactional
    async def find_by_user(self, user_id: int) -> list[Order]:
        return await self.find_all_by_field(Order.user_id, user_id)


class AuditRepo(VsDbRepo[AuditLog]):
    def __init__(self):
        super().__init__(AuditLog)


class OrderService:
    def __init__(self):
        self.order_repo = OrderRepo()
        self.audit_repo = AuditRepo()

    @transactional
    async def place_order(self, user_id: int, total: float) -> Order:
        order = await self.order_repo.create(Order(user_id=user_id, total=total, status="pending"))
        await self._log(user_id, f"order:{order.id}:placed")
        return order

    @transactional
    async def cancel_order(self, order_id: int, user_id: int) -> bool:
        result = await self.order_repo.soft_delete(order_id)
        await self._log(user_id, f"order:{order_id}:cancelled")
        return result

    @transactional(propagation=Propagation.REQUIRES_NEW)
    async def _log(self, user_id: int, action: str) -> None:
        # REQUIRES_NEW: audit log commits independently even if outer tx rolls back
        await self.audit_repo.create(AuditLog(action=action, user_id=user_id))

What vs-db eliminated here:

• No async with session.begin() anywhere
• No manual session.add() / session.commit() / session.rollback()
• No created_at / updated_at field management
• No deleted_at logic for soft delete
• Audit log independence handled by one annotation — REQUIRES_NEW