pico-sqlalchemy

Pico-SQLAlchemy

Pico-SQLAlchemy integrates Pico-IoC with SQLAlchemy, providing a true inversion of control persistence layer with Spring Data-style declarative features.

It brings constructor-based dependency injection, implicit transaction management, and powerful declarative queries using pure Python and SQLAlchemy’s Async ORM.

Requires Python 3.11+ Async-Native: Built entirely on AsyncSession and create_async_engine. Zero-Boilerplate: Repositories are transactional by default. Declarative Queries: Define SQL or expressions in decorators; the library executes them for you.

Why pico-sqlalchemy?

Most Python apps suffer from manual session handling (async with session...), scattered transaction logic, and verbose repository patterns.

Pico-SQLAlchemy solves this by offering:

Core Features

• Implicit Transactions: Methods inside @repository are automatically Read-Write transactional.
• Declarative Queries: Use @query to run SQL or Expressions automatically (defaults to Read-Only).
• AOP-Based Propagation: REQUIRED, REQUIRES_NEW, MANDATORY, NEVER, etc.
• Session Lifecycle: Centralized SessionManager handles engine creation and cleanup.
• Pagination: Built-in support for paged results via @query(paged=True).

Installation

pip install pico-sqlalchemy

You will also need an async database driver:

pip install aiosqlite   # for SQLite
pip install asyncpg     # for PostgreSQL

Quick Example

1. Define Model

from sqlalchemy import Integer, String
from pico_sqlalchemy import AppBase, Mapped, mapped_column

class User(AppBase):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    username: Mapped[str] = mapped_column(String(50))

2. Define Repository (The "Magic" Part)

Notice we don't need @transactional here.

• save: Automatically runs in a Read-Write transaction.
• find_by_name: Automatically runs in a Read-Only transaction and executes the query logic.

from pico_sqlalchemy import repository, query, SessionManager, get_session

@repository(entity=User)
class UserRepository:
    def __init__(self, manager: SessionManager):
        self.manager = manager

    # IMPLICIT: Read-Write Transaction
    async def save(self, user: User) -> User:
        session = get_session(self.manager)
        session.add(user)
        return user

    # DECLARATIVE: Read-Only Transaction + Auto-Execution
    @query(expr="username = :username", unique=True)
    async def find_by_name(self, username: str) -> User | None:
        ... # Body is ignored; the library executes the query

3. Define Service

Use @transactional here to define business logic boundaries.

from pico_ioc import component
from pico_sqlalchemy import transactional

@component
class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    @transactional
    async def create(self, name: str) -> User:
        # 1. Check existence (Read-Only tx from repo)
        existing = await self.repo.find_by_name(name)
        if existing:
            raise ValueError("User exists")
            
        # 2. Save new user (Joins current transaction)
        return await self.repo.save(User(username=name))

4. Run it

import asyncio
from pico_ioc import init, configuration, DictSource

config = configuration(DictSource({
    "database": {
        "url": "sqlite+aiosqlite:///:memory:",
        "echo": False
    }
}))

async def main():
    container = init(modules=["pico_sqlalchemy", "__main__"], config=config)
    service = await container.aget(UserService)
    
    user = await service.create("alice")
    print(f"Created: {user.id}")
    
    await container.cleanup_all_async()

if __name__ == "__main__":
    asyncio.run(main())

Transaction Hierarchy & Rules

Pico-SQLAlchemy applies a "Best Effort" strategy to determine transaction configuration. The priority order (highest wins) is:

Example Scenarios

1. Plain Method in Repository:

   async def update_user(self): ...
   

   Result: Active Read-Write Transaction (Implicit from @repository).

2. Query Method:

   @query("SELECT ...")
   async def get_data(self): ...
   

   Result: Active Read-Only Transaction (Implicit from @query).

3. Manual Override:

   @transactional(read_only=True)
   async def complex_report(self): ...
   

   Result: Active Read-Only Transaction (Explicit override).

Transaction-scoped components (v0.4.0+)

Beyond managing the SQLAlchemy session, the interceptor binds pico-ioc's "transaction" DI scope to the same boundary. A component registered with scope="transaction" is instantiated once per database transaction and torn down (running its @cleanup hooks) when that transaction ends:

@component(scope="transaction")
class UnitOfWorkAudit:
    def __init__(self):
        self.events: list[str] = []

    @cleanup
    def flush(self):
        # runs exactly when the enclosing transaction ends
        ...

A new transaction (REQUIRES_NEW, or REQUIRED with no enclosing transaction) opens a fresh scope; joins reuse the enclosing one — so the session boundary and the DI lifetime are two facets of a single transaction. Requires pico-ioc ≥ 2.2.6.

Declarative Queries in Depth

The @query decorator eliminates boilerplate for common fetches.

Expression Mode (expr)

Requires @repository(entity=Model). Injects the expression into a SELECT * FROM table WHERE ....

@query(expr="age > :min_age", unique=False)
async def find_adults(self, min_age: int) -> list[User]: ...

SQL Mode (sql)

Executes raw SQL. Useful for complex joins or specific DTOs.

@query(sql="SELECT count(*) as cnt FROM users")
async def count_users(self) -> int: ...

Automatic Pagination

Just add paged=True and a page: PageRequest parameter.

from pico_sqlalchemy import Page, PageRequest

@query(expr="active = true", paged=True)
async def find_active(self, page: PageRequest) -> Page[User]: ...

Testing

Testing is simple because you can override the configuration or the components easily using Pico-IoC.

@pytest.mark.asyncio
async def test_service():
    # Setup container with in-memory DB
    container = ... 
    
    service = await container.aget(UserService)
    user = await service.create("test")
    
    assert user.id is not None

Architecture Overview

                 ┌─────────────────────────────┐
                 │          Your App           │
                 └──────────────┬──────────────┘
                                │
                        Constructor Injection
                                │
                 ┌──────────────▼───────────────┐
                 │          Pico-IoC            │
                 └──────────────┬───────────────┘
                                │
                 ┌──────────────▼───────────────┐
                 │       pico-sqlalchemy        │
                 │ 1. Implicit Repo Transactions│
                 │ 2. Declarative @query        │
                 │ 3. Explicit @transactional   │
                 └──────────────┬───────────────┘
                                │
                           SQLAlchemy
                           (Async ORM)

AI Coding Skills

Install Claude Code or OpenAI Codex skills for AI-assisted development with pico-sqlalchemy:

curl -sL https://raw.githubusercontent.com/dperezcabrera/pico-skills/main/install.sh | bash -s -- sqlalchemy

All skills: curl -sL https://raw.githubusercontent.com/dperezcabrera/pico-skills/main/install.sh | bash

See pico-skills for details.

License

MIT