analytics-platform (Python)

One package for your Python server, run from your analytics dashboard: a firewall, error monitoring and server analytics (payments, sign-ups, leads). Django, Flask, FastAPI and any WSGI or ASGI app. No dependencies; Python 3.8 or later.

pip install analytics-platform

Django

# settings.py
MIDDLEWARE = [
    "analytics_platform.django.PlatformMiddleware",  # first
    # ...
]

ANALYTICS_PLATFORM = {
    "site_id": "YOUR_SITE_ID",
    "secret": os.environ["ANALYTICS_SITE_SECRET"],  # site settings; keep it on the server
    "endpoint": "https://YOUR_COLLECTOR_URL",
    # "trust_proxy": 1,             # behind Nginx or a load balancer
    # "trust_proxy": "cloudflare",  # every request through Cloudflare
}

In your code: from analytics_platform.django import get_platform.

FastAPI / Starlette

from analytics_platform import create_platform, PlatformMiddleware

platform = create_platform(site_id="YOUR_SITE_ID", secret=os.environ["ANALYTICS_SITE_SECRET"],
                           endpoint="https://YOUR_COLLECTOR_URL")
app.add_middleware(PlatformMiddleware, platform=platform)

Flask

platform = create_platform(site_id="YOUR_SITE_ID", secret=os.environ["ANALYTICS_SITE_SECRET"],
                           endpoint="https://YOUR_COLLECTOR_URL")
platform.init_flask(app)

Other WSGI apps: app = platform.wsgi(app); other ASGI apps: app = platform.asgi(app).

Modules

Each takes True, False or its options as a dict:

create_platform(site_id=..., secret=..., endpoint=...,
                firewall={"mode": "monitor", "cache_file": "/tmp/firewall-rules.json"},
                errors={"capture_uncaught": False},
                analytics=False)

Switch modules off in the dashboard (Security → Server package) without a deploy: the package checks about once a minute. Modules are independent: one failing to start, or the collector being down, never stops the others, and never slows your requests. Background work runs in daemon threads, which start again in each worker after a fork (gunicorn, uWSGI).

Firewall

Rules, monitor or block mode and reports are in the dashboard (Security → Firewall). New sites start in monitor mode.

Options: mode ("monitor" or "block", overrides the dashboard), cache_file (keep rules between restarts), trust_proxy, counter.

The visitor's address: behind a proxy, every request comes from the proxy unless you set trust_proxy: 1 for one proxy or load balancer, 2 for Cloudflare in front of a load balancer, "cloudflare" when every request comes through Cloudflare. (Uvicorn with --proxy-headers already does this: leave it unset.)

Rate limit with several workers: each worker counts on its own, so the limit is per worker. For one count, pass a shared counter:

def counter(key, ttl):
    hits = redis.incr(key)
    if hits == 1:
        redis.expire(key, ttl)
    return hits

create_platform(..., firewall={"counter": counter})

Bot protection: Django's failed logins are counted on their own; elsewhere call platform.login_failed(request) when a login fails. Past the site's limit the address is banned here straight away and on your other servers within a minute or two. Floods (over the rate limit) are banned the same way.

Probes. Requests for files no visitor ever asks for (/.env, /.git/…, backups: the list comes with the rules) are rule probe; in block mode the address is also banned on this server, and the collector bans it on the site's other servers.

Error monitoring

Uncaught exceptions are sent on their own; request errors through the middleware (Django, Flask, FastAPI). Errors you catch:

try:
    charge_card(order)
except PaymentError as error:
    platform.capture_exception(error, status=502)
    raise

Exceptions your code only logs: errors={"capture_logging": True} sends an exception logged at ERROR or above (logger.exception(...), or exc_info=True); it is still logged. An exception is sent once however many of these see it, and one logged inside a Django, Flask or ASGI request keeps that request's route. A batch the collector couldn't take is tried once more a few seconds later.

What still can't be sent waits in a file in the temp folder and is sent when it can be, after a restart too; each error keeps the time it happened.

Background jobs, as a decorator or a with block (sync or async):

@platform.job("send-receipts")
def send_receipts(): ...

Their errors carry the job's name as the route, a job that raises is sent (status 500) and raised on, and each run is timed (Server speed → Jobs). Several services on one site: Platform(..., server_name="worker").

Several apps on one site (microservices, a worker, a Node service next to this one): install the package in each, with the same site_id, secret and endpoint, and a server_name of its own. Errors, Server speed, Crawlers and Security → Updates then tell the apps apart. Without one, each app is named after its machine: two apps on one machine are reported as one, and only the last package list sent is kept. A service only your other apps call takes firewall=False, proxy=False, maintenance=False (its requests all come from your own apps, so the rate limit would see one visitor); on a machine shared by several apps, keep resources on in only one of them.

Options: server_name, root (traceback paths relative to it), capture_uncaught, capture_logging, spool_dir (where unsent errors wait; False for none), before_send(event, error) (change an error, or return None to drop it).

Server analytics

For what only your server knows for sure. Not pageviews: the tracker counts those.

session = platform.session_id(request)  # the visit, from the X-Analytics-Session header

platform.track("purchase", {"numeric_value": "49.90", "currency": "EUR", "event_uid": f"order-{order.id}"}, session_id=session)
platform.identify(user.id, session_id=session)
platform.lead({"email": "ana@example.com", "name": "Ana", "status": "qualified"})

track() sends in the background; identify() and lead() wait for the answer (up to 5 seconds).

Link to the visit: send the tracker's session ID with your site's requests:

fetch("/api/orders", { headers: { "x-analytics-session": window.analyticsSession?.getId() || "" }, ... });

Ad blockers

Many visitors use an ad blocker, and most blockers stop analytics. They recognise the address a script comes from and what it is called, so a tracker loaded from us is simply never counted.

This package already runs on your own domain, so it serves the tracker and takes the visits itself. Nothing to set up: use the two addresses it gives you in your page's template.

platform = get_platform()          # Django; elsewhere your own platform
context = {
    "script_url": platform.proxy.script_url(),    # /3f9a2c1b/js/v1/s.js
    "collect_url": platform.proxy.collect_url(),  # /3f9a2c1b/api/e
}

<script async src="{{ script_url }}" data-siteid="..." data-domain="example.com"
        data-collect="{{ collect_url }}"></script>

• The path comes from your secret, so it isn't in the page and isn't a name any blocker lists. Pick your own with proxy={"prefix": "..."}.
• Tracker files are fetched once and then held in memory; if we can't be reached, the copy already held is served rather than nothing.
• The visitor's address goes on as X-Forwarded-For (your collector needs it for the country, the daily visitor ID and the rate limit) and cookies never leave your server.
• Beacons are answered before the firewall, the crawler count and the timing, so your own analytics traffic doesn't appear as pages of your site. On ASGI the send runs in a thread, so the event loop keeps turning.
• Works through the Django, WSGI (Flask, Bottle, Pyramid) and ASGI (FastAPI, Starlette) middleware alike.
• Switch it off with proxy=False, or in the dashboard (Site settings → Server). Then load the tracker from us as before.

Shutting down and serverless

platform.flush() sends everything waiting (on Lambda, before returning). platform.close() also stops the background threads.