1from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
2from sqlalchemy.orm import sessionmaker
3from sqlalchemy.future import select
4from typing import AsyncGenerator
5
6DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
7
8engine = create_async_engine(DATABASE_URL, echo=True)
9AsyncSessionLocal = sessionmaker(
10 engine, class_=AsyncSession, expire_on_commit=False
11)
12
13async def get_db() -> AsyncGenerator[AsyncSession, None]:
14 async with AsyncSessionLocal() as session:
15 try:
16 yield session
17 await session.commit()
18 except Exception:
19 await session.rollback()
20 raise
21 finally:
22 await session.close()
23
24async def get_user(db: AsyncSession, user_id: int):
25 result = await db.execute(
26 select(User).where(User.id == user_id)
27 )
28 return result.scalar_one_or_none()
29
30@app.get("/users/{user_id}")
31async def read_user(user_id: int, db: AsyncSession = Depends(get_db)):
32 user = await get_user(db, user_id)
33 if user is None:
34 raise HTTPException(status_code=404, detail="User not found")
35 return user