FastAPI: Dependency injection error

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session

app = FastAPI()

# データベースセッションの依存性
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# エンドポイントで使用
@app.get("/users")
async def get_users(db: Session = Depends(get_db)):
    return db.query(User).all()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# ❌ NG: インスタンスメソッドを直接使用
class UserService:
    def get_user(self, user_id: int):
        pass

@app.get("/users/{user_id}")
async def get_user(user_id: int, service = Depends(UserService.get_user)):
    pass  # これは動かない

# ✅ OK: クラスをDependsで使用
class UserService:
    def __init__(self, db: Session = Depends(get_db)):
        self.db = db

    def get_user(self, user_id: int):
        return self.db.query(User).filter(User.id == user_id).first()

@app.get("/users/{user_id}")
async def get_user(user_id: int, service: UserService = Depends()):
    return service.get_user(user_id)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from fastapi import HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Security(security)
) -> User:
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = await get_user_by_id(user_id)
    if user is None:
        raise HTTPException(status_code=401, detail="User not found")
    return user

@app.get("/me")
async def get_me(current_user: User = Depends(get_current_user)):
    return current_user

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# 同じリクエスト内で複数回呼ばれても1回だけ実行
@app.get("/items")
async def get_items(
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
    settings: Settings = Depends(get_settings),  # get_settingsは1回だけ呼ばれる
):
    pass

# キャッシュを無効化
@app.get("/items")
async def get_items(
    value1 = Depends(get_random, use_cache=False),
    value2 = Depends(get_random, use_cache=False),  # 異なる値
):
    pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from fastapi.testclient import TestClient

def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

# 依存性を差し替え
app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_get_users():
    response = client.get("/users")
    assert response.status_code == 200

# テスト後にクリア
app.dependency_overrides.clear()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

async def get_user_or_404(
    user_id: int,  # パスパラメータ
    db: Session = Depends(get_db)
) -> User:
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.get("/users/{user_id}")
async def get_user(user: User = Depends(get_user_or_404)):
    return user

@app.put("/users/{user_id}")
async def update_user(
    user: User = Depends(get_user_or_404),
    data: UserUpdate = Body(...),
):
    pass

1
2
3
4
5
6
7
8
9

# 全エンドポイントに適用
app = FastAPI(dependencies=[Depends(verify_api_key)])

# 特定のルーターに適用
router = APIRouter(dependencies=[Depends(get_current_user)])

@router.get("/items")
async def get_items():
    pass  # 自動的にget_current_userが呼ばれる

1
2
3
4
5
6
7
8
9

# 非同期の依存性
async def get_async_db():
    async with AsyncSession() as session:
        yield session

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(select(User))
    return result.scalars().all()