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:
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:
@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:
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:
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:
{% extends "pages/base.html" %}
{% block title %}Dashboard{% endblock %}
{% block content %}
<div class="max-w-4xl mx-auto px-6 py-12">
<!-- Header -->
<div class="flex items-center justify-between mb-10">
<div>
<h1 class="text-xl font-semibold tracking-tight">Articles</h1>
<p class="text-sm text-muted mt-0.5">Manage your blog posts.</p>
</div>
<div class="flex items-center gap-3">
<a href="/dashboard/editor"
class="px-4 py-2 text-sm font-medium text-white bg-gray-900 rounded-lg hover:bg-gray-800 transition-colors">
New article
</a>
<a href="/" class="text-sm text-muted hover:text-accent transition-colors">View blog</a>
<a href="/dashboard/analytics" class="text-sm text-muted hover:text-accent transition-colors">Analytics</a>
<a href="/auth/logout" class="text-sm text-muted hover:text-accent transition-colors">Sign out</a>
</div>
</div>
<div class="grid grid-cols-1 gap-5 sm:grid-cols-2 mb-10">
<div class="overflow-hidden rounded-xl bg-white p-6 shadow-xs border border-gray-100">
<dt class="truncate text-sm font-medium text-gray-500">Total Pageviews</dt>
<dd class="mt-2 text-4xl font-semibold tracking-tight text-gray-900 font-sans">{{ total_views }}</dd>
</div>
<div class="overflow-hidden rounded-xl bg-white p-6 shadow-xs border border-gray-100">
<dt class="truncate text-sm font-medium text-gray-500">Unique Visitors (Est.)</dt>
<dd class="mt-2 text-4xl font-semibold tracking-tight text-gray-900 font-sans">-</dd>
</div>
</div>
<div class="grid grid-cols-1 gap-8 md:grid-cols-2">
<div class="rounded-xl bg-white p-6 shadow-xs border border-gray-100">
<h3 class="text-base font-semibold text-gray-900 mb-4 font-sans">Top Pages</h3>
<div class="space-y-3.5">
{% if top_pages %}
{% for page in top_pages %}
<div class="relative flex items-center justify-between py-1 px-2 group">
<div class="absolute inset-y-0 left-0 bg-orange-50 rounded-md transition-all duration-500"
style="width: {{ page.pct }}%;"></div>
<span
class="relative z-10 truncate text-sm font-medium text-gray-700 font-mono pr-4 group-hover:text-orange-700 transition-colors">
{{ page.path }}
</span>
<span
class="relative z-10 text-sm font-semibold text-gray-900 font-sans bg-white/80 px-1.5 py-0.5 rounded-md shadow-2xs border border-gray-100">
{{ page.count }}
</span>
</div>
{% endfor %}
{% else %}
<p class="text-sm text-gray-400 py-4 text-center">No pageview data found yet.</p>
{% endif %}
</div>
</div>
<div class="rounded-xl bg-white p-6 shadow-xs border border-gray-100">
<h3 class="text-base font-semibold text-gray-900 mb-4 font-sans">Top Referrers</h3>
<div class="space-y-3.5">
{% if top_referrers %}
{% for ref in top_referrers %}
<div class="relative flex items-center justify-between py-1 px-2 group">
<div class="absolute inset-y-0 left-0 bg-gray-100 rounded-md transition-all duration-500"
style="width: {{ ref.pct }}%;"></div>
<span
class="relative z-10 truncate text-sm font-medium text-gray-700 pr-4 group-hover:text-gray-900 transition-colors">
{{ ref.source }}
</span>
<span
class="relative z-10 text-sm font-semibold text-gray-900 font-sans bg-white/80 px-1.5 py-0.5 rounded-md shadow-2xs border border-gray-100">
{{ ref.count }}
</span>
</div>
{% endfor %}
{% else %}
<p class="text-sm text-gray-400 py-4 text-center">No referrer data found yet.</p>
{% endif %}
</div>
</div>
</div>
<!-- Table -->
<table class="table-auto text-xs my-10">
<thead>
<tr>
<th>Path</th>
<th>Referrer</th>
<th>User-Agent</th>
<th>Language</th>
<th>Timestamp</th>
</tr>
</thead>
<tbody>
{% for visit in visits %}
<tr class="px-1">
<td class="px-1">{{ visit.path }}</td>
<td class="px-1">{{ visit.referrer }}</td>
<td class="px-1">{{ visit.user_agent }}</td>
<td class="px-1">{{ visit.language }}</td>
<td class="px-1">{{ visit.timestamp }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% endblock %}
A little adjustment now on the router to display the figures:
@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:
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.