There is not off-the-shelf solution to capture traffic on a FastAPI application, at least not in FastAPI directly. Hopefully, doing this is pretty straightforward with FastAPI middlewares. > What is a middleware ? As per FastAPI's definition, it is a function that sits between the client and the server, that does something before the request is processed by the server, or before the response gets sent to the client. A common example is to measure the time taken by a request: ```py import time from fastapi import FastAPI, Request app = FastAPI() @app.middleware("http") async def add_process_time_header(request: Request, call_next): start_time = time.perf_counter() response = await call_next(request) process_time = time.perf_counter() - start_time response.headers["X-Process-Time"] = str(process_time) return response ``` Using it to log every request is therefore very easy: ```py @app.middleware("http") async def log_visit(request: Request, call_next): path = request.url.path if request.method == "GET" and path.startswith("/blog"): asyncio.create_task(log_user_visit( path=path, referrer=request.headers.get("referer", ""), user_agent=request.headers.get("user-agent", ""), language=request.headers.get("accept-language", "").split(",")[0], timestamp=datetime.now(timezone.utc) )) return await call_next(request) ``` **Important note here: you cannot use `BackgroundTasks` here. As a matter of fact, you cannot use any `Dependency` in a middleware. ** The `log_user_visit` itself is pretty straightfoward too: ```py async def log_user_visit(path, referrer, user_agent, language, timestamp): with Session(engine) as session: visit = Visit( path=path, referrer=referrer, user_agent=user_agent, language=language, timestamp=timestamp) session.add(visit) session.commit() ``` Since our middleware is async, we need to make sure the function here is async as well. In reality, this function is not. The writing to the DB is purely synchronous. For simple use cases, you'll be all right with that, but for larger apps you'll need to use the async versions of sqlite/postgre. I use the following Pydantic model to create my table and store my data: ```py class Visit(SQLModel, table=True): id: Optional[int] = Field(default=None, primary_key=True) path: str referrer: str = "" user_agent: str = "" language: str = "" timestamp: datetime ``` And voila. The rest is about creating a nice UI: ```html {% extends "pages/base.html" %} {% block title %}Dashboard{% endblock %} {% block content %}

Articles

Manage your blog posts.

New article View blog Analytics Sign out

Total Pageviews

Unique Visitors (Est.)

Top Pages

{% if top_pages %} {% for page in top_pages %}

{% endfor %} {% else %}

No pageview data found yet.

{% endif %}

Top Referrers

{% if top_referrers %} {% for ref in top_referrers %}

{% endfor %} {% else %}

No referrer data found yet.

{% endif %}

{% for visit in visits %} {% endfor %}

Path	Referrer	User-Agent	Language	Timestamp
{{ visit.path }}	{{ visit.referrer }}	{{ visit.user_agent }}	{{ visit.language }}	{{ visit.timestamp }}

{% endblock %} ``` A little adjustment now on the router to display the figures: ```py @router.get("/analytics") def view_analytics( request: Request, user: User = Depends(get_current_user), session: Session = Depends(get_session), ): visits = AnalyticsService.get_views(session) return templates.TemplateResponse( request=request, name="pages/analytics.html", context={ "visits": visits['raw_data'], "total_views": visits['total_views'], "top_pages": visits['top_pages'], "top_referrers": visits['top_referrers']} ) ``` And on the service: ```py def get_views(session): total_views = session.exec(select(func.count(Visit.id))).one() pages_query = select(Visit.path, func.count(Visit.id).label("count"))\ .group_by(Visit.path)\ .order_by(func.count(Visit.id).desc())\ .limit(5) top_pages = session.exec(pages_query).all() referrers_query = select(Visit.referrer, func.count(Visit.id).label("count"))\ .group_by(Visit.referrer)\ .order_by(func.count(Visit.id).desc())\ .limit(5) top_referrers = session.exec(referrers_query).all() max_page_views = top_pages[0][1] if top_pages else 1 max_ref_views = top_referrers[0][1] if top_referrers else 1 pages_data = [{"path": p, "count": c, "pct": (c / max_page_views) * 100} for p, c in top_pages] referrers_data = [{"source": r if r else "Direct / None", "count": c, "pct": (c / max_ref_views) * 100} for r, c in top_referrers] statement = select(Visit).order_by(Visit.timestamp.desc()) raw_data = session.exec(statement).all() return {"total_views": total_views, "top_pages": pages_data, "top_referrers": referrers_data, "raw_data": raw_data} ``` Thank you and good night.