@route.get("/users/{user_id}") async def get_user(user_id: int, session: AsyncSession) -> dict: # session is auto-injected (REQUEST scope) result = await session.execute(select(User).where(User.id == user_id)) return {"user": result.scalar_one_or_none()}


## Configuration

Environment variables use the `MYFY_DATA_` prefix:

| Variable | Default | Description |
|----------|---------|-------------|
| `MYFY_DATA_DATABASE_URL` | `sqlite+aiosqlite:///./myfy.db` | Database connection URL |
| `MYFY_DATA_POOL_SIZE` | `5` | Number of connections in pool |
| `MYFY_DATA_MAX_OVERFLOW` | `10` | Extra connections beyond pool_size |
| `MYFY_DATA_POOL_TIMEOUT` | `30.0` | Seconds to wait for connection |
| `MYFY_DATA_POOL_RECYCLE` | `3600` | Seconds before connection recycled |
| `MYFY_DATA_POOL_PRE_PING` | `True` | Test connections before use |
| `MYFY_DATA_ECHO` | `False` | Log all SQL statements |
| `MYFY_DATA_ENVIRONMENT` | `development` | Environment (blocks auto_create in production) |

## Supported Databases

```python
# SQLite (development)
MYFY_DATA_DATABASE_URL="sqlite+aiosqlite:///./app.db"

# PostgreSQL (production)
MYFY_DATA_DATABASE_URL="postgresql+asyncpg://user:pass@localhost/db"

# MySQL
MYFY_DATA_DATABASE_URL="mysql+aiomysql://user:pass@localhost/db"

Defining Models

from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    name: Mapped[str] = mapped_column(String(100))

Auto-Create Tables (Development Only)

from myfy.data import DataModule

app.add_module(DataModule(
    auto_create_tables=True,  # Only in development!
    metadata=Base.metadata,
))

Raises AutoCreateTablesProductionError if MYFY_DATA_ENVIRONMENT=production.

Using Sessions in Routes

Sessions are REQUEST-scoped (one per HTTP request):

from myfy.data import AsyncSession
from sqlalchemy import select

@route.get("/users")
async def list_users(session: AsyncSession) -> list[dict]:
    result = await session.execute(select(User))
    users = result.scalars().all()
    return [{"id": u.id, "name": u.name} for u in users]

@route.post("/users", status_code=201)
async def create_user(body: UserCreate, session: AsyncSession) -> dict:
    user = User(**body.model_dump())
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return {"id": user.id}

Using Sessions in Providers

from myfy.core import provider, REQUEST
from myfy.data import AsyncSession

@provider(scope=REQUEST)
def user_repository(session: AsyncSession) -> UserRepository:
    return UserRepository(session)

Transactions

Sessions auto-commit on success, rollback on exception:

@route.post("/transfer")
async def transfer(body: TransferRequest, session: AsyncSession) -> dict:
    # Both updates succeed or both rollback
    sender = await session.get(Account, body.sender_id)
    receiver = await session.get(Account, body.receiver_id)

    sender.balance -= body.amount
    receiver.balance += body.amount

    await session.commit()  # Explicit commit
    return {"success": True}

Using SessionFactory Directly

For background jobs or manual session management:

from myfy.data import SessionFactory

@provider(scope=SINGLETON)
def background_service(factory: SessionFactory) -> BackgroundService:
    return BackgroundService(factory)

class BackgroundService:
    async def process(self):
        async with self.factory.session_context() as session:
            # Manual session management
            await self.do_work(session)

Alembic Migrations

Initialize Alembic:

alembic init migrations

Configure alembic.ini:

sqlalchemy.url = driver://user:pass@localhost/dbname

Configure migrations/env.py:

from app.models import Base
target_metadata = Base.metadata

Create and run migrations:

alembic revision --autogenerate -m "Add users table"
alembic upgrade head

Best Practices

Use async drivers - asyncpg for PostgreSQL, aiosqlite for SQLite
Never share sessions - Sessions are REQUEST-scoped for a reason
Use migrations in production - Don't use auto_create_tables in production
Configure pool size - Match to expected concurrent connections
Enable pool_pre_ping - Detects stale connections before use
Use transactions - Group related operations in a single commit

data-module

DataModule - Database Access

Quick Start