Using SQLAlchemy with async Python requires understanding async patterns. Here’s how.

Setup

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Async Models

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String)

Async Operations

async def create_user(name: str):
    async with AsyncSessionLocal() as session:
        user = User(name=name)
        session.add(user)
        await session.commit()
        return user

async def get_user(user_id: int):
    async with AsyncSessionLocal() as session:
        result = await session.get(User, user_id)
        return result

Best Practices

  1. Use async context managers
  2. Commit transactions explicitly
  3. Handle exceptions properly
  4. Use connection pooling
  5. Close sessions correctly

Conclusion

Build efficient async database applications with SQLAlchemy! 🐍