SQLAlchemy 2.0 ও Alembic

uv add "sqlalchemy[asyncio]" asyncpg alembic

# app/models.py
from datetime import datetime
from sqlalchemy import BigInteger, ForeignKey, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class Customer(Base):
    __tablename__ = "customers"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    phone: Mapped[str] = mapped_column(String(15), unique=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    orders: Mapped[list["Order"]] = relationship(back_populates="customer")


class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"), index=True)
    total: Mapped[int]                                   # পয়সায়
    status: Mapped[str] = mapped_column(String(20), default="pending")
    note: Mapped[str | None]                             # None হতে পারে → NULL হতে পারে
    created_at: Mapped[datetime] = mapped_column(server_default=func.now(), index=True)

    customer: Mapped[Customer] = relationship(back_populates="orders")

# app/db.py
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

DATABASE_URL = "postgresql+asyncpg://app:secret@localhost:5432/food"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,          # একসাথে সর্বোচ্চ ১০টা খোলা connection
    max_overflow=5,        # চাপে আরও ৫টা পর্যন্ত
    pool_pre_ping=True,    # ব্যবহারের আগে connection জীবিত কিনা দেখা
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

import asyncio
from sqlalchemy import select, update, delete, func
from app.db import SessionLocal
from app.models import Customer, Order


async def main() -> None:
    async with SessionLocal() as session:
        # CREATE
        rahim = Customer(name="Rahim", phone="01711111111")
        session.add(rahim)
        await session.flush()                  # INSERT চালাও, id পাও (commit এখনো না)
        session.add(Order(customer_id=rahim.id, total=45000))
        await session.commit()                 # সব স্থায়ী

        # READ: একটা
        c = await session.get(Customer, rahim.id)

        # READ: query দিয়ে
        stmt = select(Customer).where(Customer.phone == "01711111111")
        c = await session.scalar(stmt)          # একটা বা None

        stmt = (
            select(Order)
            .where(Order.status == "pending", Order.total > 10000)
            .order_by(Order.created_at.desc())
            .limit(20)
        )
        orders = (await session.scalars(stmt)).all()

        # UPDATE: object বদলে commit
        c.name = "Rahim Uddin"
        await session.commit()

        # UPDATE: সরাসরি অনেকগুলো
        await session.execute(
            update(Order).where(Order.status == "pending").values(status="cancelled")
        )
        await session.commit()

        # DELETE
        await session.delete(c)
        await session.commit()


asyncio.run(main())

stmt = (
    select(Customer.name, func.count(Order.id), func.sum(Order.total))
    .join(Order, Order.customer_id == Customer.id)
    .where(Order.status == "delivered")
    .group_by(Customer.name)
    .order_by(func.sum(Order.total).desc())
)
for name, count, total in await session.execute(stmt):
    print(name, count, total / 100)

orders = (await session.scalars(select(Order).limit(20))).all()
for o in orders:
    print(o.customer.name)     # প্রতিটা order এ আলাদা query!

from sqlalchemy.orm import selectinload, joinedload

# পদ্ধতি ১: JOIN করে একসাথে (many-to-one এর জন্য ভালো)
stmt = select(Order).options(joinedload(Order.customer)).limit(20)

# পদ্ধতি ২: দ্বিতীয় একটা query তে সবগুলো (one-to-many এর জন্য ভালো)
stmt = select(Customer).options(selectinload(Customer.orders)).limit(20)
# SELECT ... FROM customers LIMIT 20
# SELECT ... FROM orders WHERE customer_id IN (1, 2, 3, ...)   ← মাত্র ২টা query

async with SessionLocal() as session:
    async with session.begin():               # এই block শেষে commit, error হলে rollback
        sender = await session.get(Wallet, 1, with_for_update=True)    # row lock
        receiver = await session.get(Wallet, 2, with_for_update=True)
        if sender.balance < 50000:
            raise InsufficientBalanceError(...)
        sender.balance -= 50000
        receiver.balance += 50000

uv run alembic init -t async migrations

from app.models import Base
target_metadata = Base.metadata

# model দেখে migration বানাও
uv run alembic revision --autogenerate -m "create customers and orders"

# database এ প্রয়োগ করো
uv run alembic upgrade head

# এক ধাপ পেছনে
uv run alembic downgrade -1

# কোন version এ আছি
uv run alembic current

def upgrade() -> None:
    op.add_column("orders", sa.Column("note", sa.String(), nullable=True))

def downgrade() -> None:
    op.drop_column("orders", "note")