vs-db 0.1.0

Async database layer for Viveka Sutra — SQLAlchemy sessions, transactions, and base repository

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:

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

config = VsIniConfig("config.ini")
VsDbSessionFactory(config)

config.ini:

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

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.

---

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