- 1. FastAPI 입문: 설치부터 Hello World 까지
- 2. Path 와 Query 매개변수 (URL 로 데이터 받기)
- 3. Pydantic 으로 Request Body 다루기
- 4. Response Model 과 HTTP 상태 코드
- 5. 예외 처리 (HTTPException 완벽 가이드)
- 6. 의존성 주입 (Dependency Injection) 정복
- 7. SQLAlchemy 비동기 데이터베이스
- 8. JWT 인증과 보안 (로그인부터 토큰 갱신까지)
- 9. 실시간 통신 (WebSockets & SSE)
- 10. 테스트 작성법 (Pytest 완전 가이드)
7. SQLAlchemy 비동기 데이터베이스
# 7 강. SQLAlchemy 비동기 데이터베이스
## 🎯 학습 목표
- AsyncEngine 과 AsyncSession 설정
- 비동기 CRUD 연산
- N+1 문제 해결
---
## 1. AsyncEngine 설정
```python
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"
engine = create_async_engine(
DATABASE_URL,
pool_pre_ping=True,
)
AsyncSessionLocal = sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
)
```
---
## 2. 비동기 CRUD
```python
@app.post("/users", response_model=User)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
db_user = User(**user.model_dump())
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return db_user
```
---
## 3. N+1 문제 해결
```python
from sqlalchemy.orm import selectinload
# ✅ selectinload 로 2 번 쿼리
result = await db.execute(
select(User).options(selectinload(User.posts))
)
users = result.scalars().unique().all()
```