SQLAlchemy 2.0 async dùng create_async_engine, async_sessionmaker, và ORM style mới với Mapped/mapped_column — fully type-safe.
python
from sqlalchemy.ext.asyncio import (
create_async_engine, AsyncSession, async_sessionmaker
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
engine = create_async_engine(
"postgresql+asyncpg://user:pass@localhost/db",
pool_size=10, max_overflow=20,
)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase): pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(unique=True, index=True)
name: Mapped[str] = mapped_column(nullable=False)
async def get_db():
async with AsyncSessionLocal() as session:
yield session