sqlalchemy-crud-tx 1.0.0

Use SQLAlchemy in a Pythonic way.

sqlalchemy-crud-tx

English | 中文

A lightweight CRUD + transaction helper for SQLAlchemy (Flask glue can be added via extensions):

• Context-managed CRUD with nested savepoints: with CRUD(Model) as crud:
• Function-level transactions via @CRUD.transaction() with join semantics
• Configurable error policy (error_policy="raise"|"status_only") and pluggable logger
• Type-friendly CRUDQuery wrapper for common chainable operations

Install

pip install sqlalchemy-crud-tx
# with Flask integration
pip install "sqlalchemy-crud-tx[flask]"
# or local editable install
pip install -e .

Requires Python 3.11+ with sqlalchemy>=1.4 (optional Flask integration can be added separately).

Quick Start (pure SQLAlchemy)

from sqlalchemy import String, Integer, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
from sqlalchemy_crud_tx import CRUD

engine = create_engine("sqlite:///./crud_example.db", echo=False)
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "example_user"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)


Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

CRUD.configure(session_provider=SessionLocal, error_policy="raise")

with CRUD(User) as crud:
    user = crud.add(email="[e-mail]")
    print("created", user)

with CRUD(User, email="[e-mail]") as crud:
    row = crud.first()
    print("fetched", row)

with CRUD(User) as crud:
    crud.delete(row)
# or
with CRUD(User, email="[e-mail]") as crud:
    crud.delete()

Function-Level Transactions

from sqlalchemy_crud_tx import CRUD

CRUD.configure(session_provider=SessionLocal, error_policy="raise")

@CRUD.transaction(error_policy="raise")
def create_two_users():
    with CRUD(User) as crud1:
        crud1.add(email="[e-mail]")
    with CRUD(User) as crud2:
        crud2.add(email="[e-mail]")

create_two_users()

• The outermost call commits or rolls back; inner CRUD contexts only mark status when exceptions occur.
• With error_policy="status_only", SQLAlchemyError is rolled back and caught; check crud.status / crud.error instead.

Docs & Examples

• Full example: docs/examples/basic_crud.py
• Transaction refactor notes/TODO: docs/crud_refactor_todo.md
• Typing directions: docs/todo.md

Testing

1. Provide a DB URI via env or .env: TEST_DB=sqlite:///./test.db (or another driver).
2. Install test deps, then:

   pytest -q
   

Notes

• SQLAlchemy-first; optional Flask integration can be layered via extensions.
• Always call CRUD.configure(session_provider=...) before using CRUD instances.
• If a Session may already be in a transaction (e.g., AUTOBEGIN after expire_on_commit), set existing_txn_policy in CRUD.configure(...) to control how CRUD behaves (error, join, savepoint, adopt_autobegin, reset).