Dependency Injection
DI แก้ปัญหาอะไร?
หัวข้อที่มีชื่อว่า “DI แก้ปัญหาอะไร?”ทุก API ล้วนมีงานที่ต้องใช้ร่วมกัน เช่น ขอ DB session, อ่าน current user จาก token หรือตรวจ query parameter ชุดเดิม ๆ ฝั่ง Express แก้ด้วย middleware ส่วน NestJS ใช้ IoC container คู่กับ provider ที่ติด @Injectable()
FastAPI เลือกอีกทาง คือใช้ Depends() เป็นระบบ DI ที่เบาและ compose ต่อกันได้ โดยประกาศทุกอย่างผ่าน type hints
Depends() vs Express middleware vs NestJS providers
หัวข้อที่มีชื่อว่า “Depends() vs Express middleware vs NestJS providers”// Express middleware (global side-effect style)function authMiddleware(req, res, next) { const token = req.headers.authorization?.split(' ')[1]; if (!token) return res.status(401).json({ error: 'Unauthorized' }); req.user = verifyToken(token); // แนบกับ req next();}app.use(authMiddleware);
// NestJS — @Injectable() provider@Injectable()export class AuthService { getCurrentUser(token: string): User { ... }}// inject ผ่าน constructorconstructor(private authService: AuthService) {}# FastAPI — Depends()from fastapi import FastAPI, Depends, HTTPException, Headerfrom typing import Optional
app = FastAPI()
# dependency คือ callable ธรรมดาdef get_current_user(authorization: Optional[str] = Header(None)): if not authorization or not authorization.startswith("Bearer "): raise HTTPException(status_code=401, detail="Unauthorized") token = authorization.removeprefix("Bearer ") return {"user_id": 1, "token": token} # return ค่าที่ต้องการ
# Inject ด้วย Depends() — รันก่อน handler@app.get("/profile")async def get_profile(current_user: dict = Depends(get_current_user)): return {"user": current_user}Reusable query parameters
หัวข้อที่มีชื่อว่า “Reusable query parameters”รูปแบบที่พบบ่อยมากคือ shared “pagination” dependency ที่หลาย route ใช้ร่วมกัน เทียบเท่ากับ custom Express middleware หรือ NestJS PaginationDto
// NestJS — shared DTOexport class PaginationDto { @IsOptional() @Type(() => Number) @IsInt() @Min(0) skip?: number = 0;
@IsOptional() @Type(() => Number) @IsInt() @Max(100) limit?: number = 10;}
@Get()findAll(@Query() pagination: PaginationDto) { ... }# FastAPI — pagination เป็น Dependsfrom fastapi import FastAPI, Depends
app = FastAPI()
class PaginationParams: def __init__(self, skip: int = 0, limit: int = 10): if skip < 0: raise HTTPException(status_code=400, detail="skip must be >= 0") if limit > 100: raise HTTPException(status_code=400, detail="limit must be <= 100") self.skip = skip self.limit = limit
@app.get("/items")async def list_items(pagination: PaginationParams = Depends()): return {"skip": pagination.skip, "limit": pagination.limit}
@app.get("/users")async def list_users(pagination: PaginationParams = Depends()): return {"skip": pagination.skip, "limit": pagination.limit}DB session dependency
หัวข้อที่มีชื่อว่า “DB session dependency”dependency ที่สำคัญที่สุดใน real app คือ database session Depends() ของ FastAPI จัดการ lifecycle — เปิด, yield ให้ handler, ปิดเมื่อเสร็จ
// NestJS — TypeORM via constructor injection@Injectable()export class ItemsService { constructor( @InjectRepository(Item) private readonly itemRepo: Repository<Item>, ) {} async findAll() { return this.itemRepo.find(); }}# FastAPI — SQLAlchemy session via Dependsfrom fastapi import FastAPI, Dependsfrom sqlalchemy.orm import Sessionfrom app.database import SessionLocal
app = FastAPI()
# Generator dependency — yield session, cleanup หลังdef get_db(): db = SessionLocal() try: yield db # ← handler ได้รับค่านี้ finally: db.close() # ← รันเสมอ แม้จะมี exception
@app.get("/items/{item_id}")async def get_item(item_id: int, db: Session = Depends(get_db)): item = db.get(Item, item_id) if not item: raise HTTPException(status_code=404, detail="Item not found") return itemการ compose dependencies
หัวข้อที่มีชื่อว่า “การ compose dependencies”Dependencies สามารถ depend บน dependencies อื่นได้ ก่อตั้งเป็น chain นี่คือวิธีที่คุณ layer auth → permissions → resource fetching
// NestJS guard chain@UseGuards(AuthGuard, RolesGuard)@Roles('admin')@Get(':id')findOne(@Param('id') id: string) { ... }# FastAPI — dependencies compose ได้ตามธรรมชาติfrom fastapi import FastAPI, Depends
app = FastAPI()
def get_current_user(token: str = Depends(get_token)): return decode_token(token)
def require_admin(user: dict = Depends(get_current_user)): if user["role"] != "admin": raise HTTPException(status_code=403, detail="Admin only") return user
@app.delete("/items/{item_id}")async def delete_item(item_id: int, _: dict = Depends(require_admin)): return {"deleted": item_id}รันที่เครื่องตัวเอง — ต้องการ FastAPI + server Playground ข้ามในบทเรียนนี้เพราะ
Depends()ต้องการ FastAPI request lifecycle ในการทำงาน