· FastAPI· dependency injection· Python· testing· SQLModel· Redis· auth· SaaS
Dependency injection i FastAPI — mønstre for SaaS-applikationer
FastAPI dependency injection: Depends(), multi-level dependencies, testbarhed, database sessions, Redis, auth middleware og performance med cached dependencies.
Af M-Bus Gateway
FastAPI's Depends() system er en af dens stærkeste funktioner. Her er avancerede mønstre til SaaS-applikationer.
Grundprincipper
# server/src/core/deps.py
from fastapi import Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from redis.asyncio import Redis
# Dependency = callable der returnerer en værdi (eller yield'er)
# FastAPI resolver automatisk afhængighedsgraf
async def get_session() -> AsyncSession:
"""Database session — yields, så cleanup sker automatisk."""
async with async_session_factory() as session:
yield session
async def get_redis() -> Redis:
"""Redis forbindelse fra connection pool."""
return redis_client
# Brug i endpoint:
@router.get("/")
async def my_endpoint(
session: AsyncSession = Depends(get_session),
redis: Redis = Depends(get_redis),
):
...
Auth dependency-kæde
# server/src/auth/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
security = HTTPBearer()
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
session: AsyncSession = Depends(get_session),
) -> TokenPayload:
"""Valider JWT og returnér decoded payload."""
try:
payload = jwt.decode(
credentials.credentials,
settings.JWT_SECRET_KEY,
algorithms=["HS256"],
)
return TokenPayload(**payload)
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Ugyldig token",
headers={"WWW-Authenticate": "Bearer"},
)
def require_role(*roles: str):
"""Factory: returnerer dependency der kræver en bestemt rolle."""
async def check_role(
user: TokenPayload = Depends(get_current_user),
) -> TokenPayload:
if user.role not in roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Kræver rolle: {', '.join(roles)}",
)
return user
return check_role
# Brug:
@router.post("/properties")
async def create_property(
user: TokenPayload = Depends(require_role("landlord", "super_admin")),
): ...
@router.get("/admin/tenants")
async def list_tenants(
user: TokenPayload = Depends(require_role("super_admin")),
): ...
Tenant-isolation dependency
# server/src/core/deps.py
async def get_tenant_property(
property_id: UUID,
user: TokenPayload = Depends(get_current_user),
session: AsyncSession = Depends(get_session),
) -> Property:
"""
Hent ejendom og verificér tenant-tilhørsforhold.
Returnerer 404 (ikke 403) for at undgå information disclosure.
"""
query = select(Property).where(
Property.id == property_id,
Property.deleted_at.is_(None),
)
# super_admin kan se alle ejendomme
if user.role != "super_admin":
query = query.where(Property.tenant_id == user.tenant_id)
result = await session.execute(query)
prop = result.scalar_one_or_none()
if not prop:
raise HTTPException(status_code=404, detail="Ejendom ikke fundet")
return prop
# Brug i endpoint:
@router.get("/properties/{property_id}/units")
async def get_units(
property: Property = Depends(get_tenant_property),
session: AsyncSession = Depends(get_session),
):
# property er allerede valideret og tenant-isoleret
units = await _get_units(session, property.id)
return units
Cached dependencies
# server/src/core/deps.py
from functools import lru_cache
from fastapi import Depends
@lru_cache(maxsize=1)
def get_settings() -> Settings:
"""Settings er immutable — caches for hele applikationens levetid."""
return Settings()
# Dyr initialisering — én gang pr. request (ikke pr. dependency-brug):
async def get_stripe_client(
settings: Settings = Depends(get_settings),
) -> stripe.StripeClient:
"""Stripe client deles på tværs af dependencies i samme request."""
return stripe.StripeClient(api_key=settings.STRIPE_SECRET_KEY)
# FastAPI deduplicerer automatisk — get_session() kaldes KUN én gang
# selvom det bruges af 3 dependencies i samme request:
@router.post("/complex-endpoint")
async def complex(
user: TokenPayload = Depends(get_current_user), # bruger get_session
property: Property = Depends(get_tenant_property), # bruger get_session
session: AsyncSession = Depends(get_session), # samme instans!
):
...
Pagination dependency
# server/src/core/deps.py
from dataclasses import dataclass
from fastapi import Query
@dataclass
class PaginationParams:
page: int
per_page: int
offset: int
@property
def limit(self) -> int:
return self.per_page
async def get_pagination(
page: int = Query(default=1, ge=1, description="Sidenummer"),
per_page: int = Query(default=20, ge=1, le=100, description="Resultater pr. side"),
) -> PaginationParams:
return PaginationParams(
page=page,
per_page=per_page,
offset=(page - 1) * per_page,
)
# Brug:
@router.get("/properties")
async def list_properties(
pagination: PaginationParams = Depends(get_pagination),
user: TokenPayload = Depends(get_current_user),
session: AsyncSession = Depends(get_session),
):
query = (
select(Property)
.where(Property.tenant_id == user.tenant_id)
.offset(pagination.offset)
.limit(pagination.limit)
)
...
Test-overrides
# server/tests/conftest.py
async def override_get_session():
"""Test-session med rollback."""
async with test_session_factory() as session:
yield session
async def override_get_current_user():
"""Hardkodet testbruger — ingen JWT parsing."""
return TokenPayload(
sub="test-user-id",
tenant_id="550e8400-e29b-41d4-a716-446655440000",
role="landlord",
)
@pytest.fixture
async def client():
app.dependency_overrides = {
get_session: override_get_session,
get_current_user: override_get_current_user,
}
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
app.dependency_overrides.clear()
# Fordel: Test-fixtures er minimale — kun de nødvendige overrides
# Resten af dependency-grafen kører uændret (inkl. pagination, validation osv.)
Rate limiting dependency
# server/src/middleware/rate_limit.py
async def rate_limit(
request: Request,
redis: Redis = Depends(get_redis),
limit: int = 100,
window: int = 60,
) -> None:
"""Sliding window rate limiter som dependency."""
ip = request.client.host
key = f"rate:{ip}:{request.url.path}"
now = time.time()
pipe = redis.pipeline()
pipe.zremrangebyscore(key, 0, now - window)
pipe.zadd(key, {str(now): now})
pipe.zcard(key)
pipe.expire(key, window)
results = await pipe.execute()
if results[2] > limit:
raise HTTPException(
status_code=429,
detail=f"Rate limit: {limit} requests/{window}s",
headers={"Retry-After": str(window)},
)
# Anvend på specifikke endpoints:
@router.post("/auth/login")
async def login(
body: LoginIn,
_: None = Depends(rate_limit), # 100 req/min pr. IP
): ...
# Eller lavere limit for dyre operationer:
from functools import partial
strict_rate_limit = partial(rate_limit, limit=10, window=60)
@router.post("/settlements/generate")
async def generate_settlement(
_: None = Depends(strict_rate_limit),
): ...
Konklusion
FastAPI's Depends() bygger automatisk en dependency-graf og deduplicerer dependencies pr. request. require_role() factory-pattern giver type-sikker RBAC. Tenant-isolation dependency returnerer 404 (ikke 403) for at undgå information disclosure. Test-overrides gør det trivielt at erstatte infrastruktur-dependencies med test-doubles.