Skip to content

Performance & Caching

Redis Caching (Cache-Aside Pattern)

from redis.asyncio import Redis


class UserService:
    def __init__(self, repo: UserRepository, redis: Redis):
        self.repo = repo
        self.redis = redis

    async def get_user(self, user_id: UUID) -> User:
        cache_key = f"user:{user_id}"

        cached = await self.redis.get(cache_key)
        if cached:
            return UserResponse.model_validate_json(cached)

        user = await self.repo.get(user_id)
        if not user:
            raise NotFoundError("User", user_id)

        await self.redis.setex(cache_key, 3600, UserResponse.model_validate(user).model_dump_json())
        return user

    async def update_user(self, user_id: UUID, data: UserUpdate) -> User:
        user = await self.repo.update(user_id, data.model_dump(exclude_unset=True))
        await self.redis.delete(f"user:{user_id}")  # Invalidate cache
        return user

Query Optimization

Select only needed columns for list endpoints:

# Full ORM objects for detail views
stmt = select(User).where(User.id == user_id)

# Column projection for list views
stmt = select(User.id, User.name, User.email).where(User.is_active.is_(True))

Cursor-based pagination for large datasets:

# Offset pagination — simple but slow for deep pages
stmt = select(User).offset((page - 1) * size).limit(size)

# Cursor pagination — consistent performance at any depth
stmt = (
    select(User)
    .where(User.created_at < cursor_timestamp)
    .order_by(User.created_at.desc())
    .limit(size)
)

Parallel I/O

Use asyncio.gather when fetching independent data:

async def get_dashboard(self, user_id: UUID) -> Dashboard:
    stats, notifications, activity = await asyncio.gather(
        self.stats_service.get_user_stats(user_id),
        self.notification_service.get_unread(user_id),
        self.activity_service.get_recent(user_id),
    )
    return Dashboard(stats=stats, notifications=notifications, activity=activity)

See Also