Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
from sqlalchemy import func
# Count
stmt = select(func.count(User.id)).where(User.is_active == True)
count = session.execute(stmt).scalar()
# Group by
stmt = (
select(User.id, func.count(Post.id).label("post_count"))
.outerjoin(Post)
.group_by(User.id)
.having(func.count(Post.id) > 5)
)
Repository Pattern
from typing import TypeVar, Genericfrom sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
T = TypeVar("T", bound=Base)
classBaseRepository(Generic[T]):
def__init__(self, db: AsyncSession, model: type[T]):
self.db = db
self.model = model
asyncdefget_by_id(self, id: int) -> T | None:
returnawaitself.db.get(self.model, id)
asyncdefget_all(self, skip: int = 0, limit: int = 100) -> list[T]:
stmt = select(self.model).offset(skip).limit(limit)
result = awaitself.db.execute(stmt)
returnlist(result.scalars().all())
asyncdefcreate(self, **kwargs) -> T:
obj = self.model(**kwargs)
self.db.add(obj)
awaitself.db.commit()
awaitself.db.refresh(obj)
return obj
asyncdefupdate(self, obj: T, **kwargs) -> T:
for key, value in kwargs.items():
setattr(obj, key, value)
awaitself.db.commit()
awaitself.db.refresh(obj)
return obj
asyncdefdelete(self, obj: T) -> None:
awaitself.db.delete(obj)
awaitself.db.commit()
classUserRepository(BaseRepository[User]):
def__init__(self, db: AsyncSession):
super().__init__(db, User)
asyncdefget_by_email(self, email: str) -> User | None:
stmt = select(User).where(User.email == email)
result = awaitself.db.execute(stmt)
return result.scalar_one_or_none()
Query Optimization
N+1 Problem
# BAD: N+1 queries
users = session.execute(select(User)).scalars().all()
for user in users:
print(user.posts) # Each access triggers a query!# GOOD: Eager loading
stmt = select(User).options(selectinload(User.posts))
users = session.execute(stmt).scalars().all()
for user in users:
print(user.posts) # No additional queries
engine = create_engine(
DATABASE_URL,
pool_size=5, # Maintained connections
max_overflow=10, # Extra connections when needed
pool_timeout=30, # Seconds to wait for connection
pool_recycle=1800, # Recycle connections after 30 min
pool_pre_ping=True, # Test connections before use
)
FastAPI Integration
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
asyncdefget_db() -> AsyncGenerator[AsyncSession, None]:
asyncwith AsyncSessionLocal() as session:
try:
yield session
finally:
await session.close()
@app.get("/users/{user_id}")asyncdefget_user(user_id: int, db: AsyncSession = Depends(get_db)):
user = await db.get(User, user_id)
ifnot user:
raise HTTPException(status_code=404)
return user
