python 55 lines · 8 steps

A background session cleaner in FastAPI

Run a periodic expired-session purge as a background task tied to the FastAPI app lifespan.

Explained by highlit

FastAPI Intermediate background-tasks asyncio lifespan graceful-shutdown sqlalchemy

1import asyncio
2import logging
3from contextlib import asynccontextmanager
4from datetime import datetime, timedelta, timezone
5 
6from fastapi import FastAPI
7 
8from .database import async_session
9from .models import Session as UserSession
10from sqlalchemy import delete
11 
12logger = logging.getLogger(__name__)
13 
14CLEANUP_INTERVAL = 300
15SESSION_TTL = timedelta(hours=24)
16 
17 
18async def purge_expired_sessions() -> int:
19    cutoff = datetime.now(timezone.utc) - SESSION_TTL
20    async with async_session() as db:
21        result = await db.execute(
22            delete(UserSession).where(UserSession.last_seen_at < cutoff)
23        )
24        await db.commit()
25        return result.rowcount
26 
27 
28async def cleanup_loop() -> None:
29    while True:
30        try:
31            await asyncio.sleep(CLEANUP_INTERVAL)
32            removed = await purge_expired_sessions()
33            if removed:
34                logger.info("Purged %d expired sessions", removed)
35        except asyncio.CancelledError:
36            logger.info("Cleanup loop stopping")
37            raise
38        except Exception:
39            logger.exception("Session cleanup failed; retrying next tick")
40 
41 
42@asynccontextmanager
43async def lifespan(app: FastAPI):
44    task = asyncio.create_task(cleanup_loop())
45    try:
46        yield
47    finally:
48        task.cancel()
49        try:
50            await task
51        except asyncio.CancelledError:
52            pass
53 
54 
55app = FastAPI(lifespan=lifespan)

01 / 01

STEP 01

Walkthrough

Space play ←→ step click any line

Three takeaways

1A lifespan context manager is the clean place to start and stop background work alongside the app.
2Cancelling a task and awaiting it lets the coroutine unwind its cleanup before shutdown completes.
3Catching broad exceptions inside the loop keeps one bad tick from killing the whole worker.

Related explainers

python

from flask import Blueprint, jsonify, request, abort
 
v1 = Blueprint("users_v1", __name__)
v2 = Blueprint("users_v2", __name__)

Versioning a Flask API with Blueprints

api versioning blueprints rest

Intermediate 7 steps

python

import time
import threading
from enum import Enum
from functools import wraps

Building a circuit breaker in Python

circuit-breaker resilience decorators

Advanced 7 steps

python

from django.contrib.postgres.search import SearchQuery, SearchRank, SearchVector
from django.core.cache import cache
from django.db.models import F
from django.http import JsonResponse

Ranked full-text search in Django

full-text-search debouncing caching

Advanced 9 steps

python

import os
from datetime import timedelta

Class-based config in a Flask app factory

configuration app-factory inheritance

Intermediate 7 steps

python

import time
 
from flask import Flask, g, request

Timing requests with Flask hooks

middleware request-lifecycle instrumentation

Intermediate 5 steps

typescript

import { Injectable, Logger, OnApplicationShutdown } from '@nestjs/common';
import { InjectRedis } from '@nestjs-modules/ioredis';
import Redis from 'ioredis';
import { DataSource } from 'typeorm';