Why FastAPI geolocation middleware is the wrong tool

Search for FastAPI geolocation middleware and you'll get the same answer every time: subclass BaseHTTPMiddleware, call an IP lookup API, stash the result on request.state.geo. It works. It's also the wrong tool, and the reason is boring rather than clever.

Middleware runs before routing. It cannot know which endpoint matched, so it fires on everything: /health, /metrics, /docs, /openapi.json, your static mounts, and every CORS preflight OPTIONS. You've just put a network call in front of your liveness probe.

A dependency runs after routing, only where you declare it. That one ordering difference fixes the problem and brings typing and testability along with it.

TL;DR

• BaseHTTPMiddleware runs on every request including health checks and docs routes. Depends() runs only on routes that ask for it.
• Fix client IP resolution first. request.client.host is your load balancer, and X-Forwarded-For is caller-controlled unless you configure Uvicorn to trust it.
• Return a Pydantic model from the dependency so the route signature shows what it gets and your editor can autocomplete it.
• Keep geolocation and risk as two separate dependencies. Cheap routes take geo, sensitive routes take both.
• Fail open on lookup errors, cache in Redis, and swap the whole thing out in tests with app.dependency_overrides.

You'll end with two composable dependencies, GeoContext and RiskProfile, that any route can pull in by adding one parameter. Roughly 120 lines total, Redis-cached, and testable without mocking HTTP.

Why FastAPI geolocation middleware is the wrong tool

The routing-order problem is the one that actually bites. Starlette runs middleware in the ASGI stack before the router resolves a path, so a geolocation middleware has no way to say "skip this for /health". You end up with a hand-maintained path prefix blocklist inside the middleware, which drifts the moment someone adds a route.

Latency is the part people feel in production. A Kubernetes liveness probe hits /health every few seconds. If your middleware makes an outbound API call on every request, that probe now depends on a third-party network round trip. When the upstream gets slow, probes time out, and your orchestrator restarts a pod that was perfectly healthy. I've watched a team spend most of a day on that one.

Then there's typing. request.state.geo is a bag with no schema. Your editor can't complete it, your reviewer can't see it in the function signature, and a typo in request.state.geo.country_code2 fails at runtime in whichever route nobody tested.

BaseHTTPMiddleware also has its own reputation for edge-case behaviour around exceptions, streaming responses, and background tasks. Plenty of people run it happily. It's still more machinery than this job needs.

What a dependency gives you instead

FastAPI's dependency injection system gives you four things, in the order you'll care about them. Per-route opt-in, so /health stays a pure function. A typed return value that shows up in the signature. Automatic caching within a single request, so declaring the same dependency twice doesn't call the API twice. And dependency_overrides, which makes testing a three-line fixture instead of a mock HTTP layer.

The FastAPI dependency docs cover the mechanics well. What they don't cover is the case for using them where your instinct says middleware, which is most of what follows.

Get the client IP right first

Everything downstream is worthless if you geolocate the wrong address. request.client.host gives you the peer that opened the TCP connection. Behind nginx, an ALB, Cloudflare, or Railway's Envoy layer, that peer is the proxy, and you'll cheerfully look up your own infrastructure for every visitor.

The fix is not to parse X-Forwarded-For in your application code. That header is whatever the caller typed unless something upstream has overwritten it, so trusting the leftmost entry hands any user the ability to claim any country they like. This is the single most common mistake in the geolocation middleware examples floating around.

Let Uvicorn do it. Its proxy header settings rewrite request.client.host from the forwarded headers, but only for peers you explicitly trust:

uvicorn app.main:app \
  --proxy-headers \
  --forwarded-allow-ips="10.0.0.0/8"

The default for --forwarded-allow-ips is 127.0.0.1, which is not what you want the moment your proxy lives on another host. Set it to your load balancer's actual range and nothing else. With that in place, request.client.host is trustworthy and the dependency stays short:

# app/deps/client_ip.py
import ipaddress
from fastapi import Request


def get_client_ip(request: Request) -> str | None:
    """The caller's public IP, or None when we don't have one worth looking up."""
    client = request.client
    if client is None:
        # Some ASGI transports (including parts of the test client) omit this.
        return None

    try:
        parsed = ipaddress.ip_address(client.host)
    except ValueError:
        return None

    # Private, loopback and reserved addresses have no public geolocation.
    # Returning None here saves a pointless round trip on every local request.
    if parsed.is_private or parsed.is_loopback or parsed.is_reserved:
        return None

    return client.host

Returning None rather than raising is deliberate. A missing IP is a normal condition, not an error, and the callers downstream all handle None already.

Running it locally

Every request from your laptop arrives as 127.0.0.1, so get_client_ip returns None and the geo dependency short-circuits before it ever calls out. That's the behaviour you want, and it means local development costs nothing.

When you do want real data in development, hardcode a test IP behind an environment flag rather than pointing the lookup at a private address. IP geolocation APIs reject private and bogon ranges, and you'll spend twenty minutes debugging an error response that was correct all along.

The geolocation dependency

One call, one model. Here's the request shape:

curl -s 'https://api.ipgeolocation.io/v3/ipgeo?apiKey=API_KEY&ip=91.128.103.196'

And the full response, which is worth reading once before you decide what to keep:

{
  "ip": "91.128.103.196",
  "location": {
    "continent_code": "EU",
    "continent_name": "Europe",
    "country_code2": "SE",
    "country_code3": "SWE",
    "country_name": "Sweden",
    "country_name_official": "Kingdom of Sweden",
    "country_capital": "Stockholm",
    "state_prov": "Stockholms län",
    "state_code": "SE-AB",
    "district": "Stockholm",
    "city": "Stockholm",
    "zipcode": "164 40",
    "latitude": "59.40510",
    "longitude": "17.95510",
    "is_eu": true,
    "country_flag": "https://ipgeolocation.io/static/flags/se_64.png",
    "geoname_id": "9972319",
    "country_emoji": "🇸🇪"
  },
  "country_metadata": {
    "calling_code": "+46",
    "tld": ".se",
    "languages": ["sv-SE", "se", "sma", "fi-SE"]
  },
  "currency": { "code": "SEK", "name": "Swedish Krona", "symbol": "kr" },
  "asn": {
    "as_number": "AS1257",
    "organization": "Tele2 Sverige AB",
    "country": "SE"
  },
  "time_zone": {
    "name": "Europe/Stockholm",
    "offset": 1,
    "offset_with_dst": 2,
    "current_time": "2026-09-07 16:55:30.494+0200",
    "current_time_unix": 1788792930.494,
    "current_tz_abbreviation": "CEST",
    "current_tz_full_name": "Central European Summer Time",
    "is_dst": true
  }
}

Two things in there will trip you up. latitude and longitude are strings, not floats, so cast them before any arithmetic. And country_metadata.languages is an array, not the comma-separated string that several other providers return. If you're porting code from another API, that's where it breaks.

I'm using ipgeolocation.io for these examples because the geolocation and threat endpoints share a response envelope, which keeps the two dependencies below nearly identical. IPGeolocation, ipinfo, ip-api, MaxMind GeoIP2, IPLocate and IP2Location all hand back roughly the same country-level payload, so swap in whichever is already in your stack. Only the parsing function changes.

Now the dependency:

# app/deps/geo.py
import logging
import os
from typing import Annotated

import httpx
from fastapi import Depends, Request
from pydantic import BaseModel

from app.deps.client_ip import get_client_ip

logger = logging.getLogger(__name__)


class GeoContext(BaseModel):
    ip: str
    country_code: str | None = None
    country_name: str | None = None
    city: str | None = None
    is_eu: bool = False
    currency_code: str | None = None
    asn_organization: str | None = None
    timezone: str | None = None


async def get_geo(
    request: Request,
    ip: Annotated[str | None, Depends(get_client_ip)],
) -> GeoContext | None:
    if ip is None:
        return None

    api_key = os.environ.get("IPGEO_API_KEY")
    if not api_key:
        logger.warning("IPGEO_API_KEY is not set, skipping geolocation")
        return None

    try:
        response = await request.app.state.http.get(
            "/ipgeo", params={"apiKey": api_key, "ip": ip}
        )
        response.raise_for_status()
        payload = response.json()
    except (httpx.HTTPError, ValueError) as exc:
        # Fail open. A geo lookup should never be why a page returns 500.
        logger.warning("Geolocation lookup failed for %s: %s", ip, exc)
        return None

    location = payload.get("location") or {}
    return GeoContext(
        ip=payload.get("ip", ip),
        country_code=location.get("country_code2"),
        country_name=location.get("country_name"),
        city=location.get("city"),
        is_eu=bool(location.get("is_eu", False)),
        currency_code=(payload.get("currency") or {}).get("code"),
        asn_organization=(payload.get("asn") or {}).get("organization"),
        timezone=(payload.get("time_zone") or {}).get("name"),
    )

The or {} on every nested access is not paranoia. Responses vary by key configuration, and payload["location"]["city"] on an IP with no city resolution is a KeyError in production at 3am.

Fail open or fail closed

The code above fails open: if the lookup breaks, get_geo returns None and the route carries on with whatever default it has. For pricing, currency, or language, that's obviously right. Nobody should see a 500 because a third-party API had a bad minute.

Fail closed is the correct choice in exactly one situation, which is when the lookup is a control rather than a decoration. If you're blocking sanctioned jurisdictions, "we couldn't check" has to mean "denied", because a fail-open geoblock is a geoblock that stops working the moment someone can make your API call time out. Decide which one each route needs and write it down, because the default you pick silently becomes policy everywhere.

Adding risk data as a second dependency

Country tells you where someone claims to be. It says nothing about whether they're behind a VPN, a residential proxy, or a datacenter. For a signup or checkout route you usually want both, and the IP Security API returns the threat side as its own endpoint:

curl -s 'https://api.ipgeolocation.io/v3/security?apiKey=API_KEY&ip=145.223.7.7'

{
  "ip": "145.223.7.7",
  "security": {
    "threat_score": 90,
    "is_tor": false,
    "is_proxy": true,
    "proxy_provider_names": ["NetNut", "ProxyScrape", "Oxy Labs", "DataImpulse"],
    "proxy_confidence_score": 99,
    "proxy_last_seen": "2026-09-01",
    "is_residential_proxy": true,
    "is_vpn": true,
    "vpn_provider_names": ["SurfShark VPN", "Ishaan VPN"],
    "vpn_confidence_score": 99,
    "vpn_last_seen": "2026-07-31",
    "is_relay": false,
    "relay_provider_name": "",
    "is_anonymous": true,
    "is_known_attacker": true,
    "is_bot": false,
    "bot_confidence_score": 0,
    "bot_operator_name": "",
    "bot_type": "",
    "is_known_good_bot": false,
    "bot_last_seen": "",
    "is_spam": true,
    "is_cloud_provider": true,
    "cloud_provider_name": "Brander Group Inc.",
    "is_corporate_gateway": false,
    "corporate_gateway_type": "",
    "corporate_gateway_provider_name": ""
  }
}

Named providers are the part worth keeping. A boolean tells you the IP is a proxy; ["NetNut", "ProxyScrape", "Oxy Labs", "DataImpulse"] tells your fraud team which network it belongs to, which is the difference between an alert and an investigation. is_corporate_gateway matters for the opposite reason: a shared Zscaler or Netskope egress address looks like a proxy by every other measure, and blocking it means blocking an entire enterprise customer.

The dependency mirrors the geo one closely enough that I'll only show what differs:

# app/deps/risk.py
import logging
import os
from typing import Annotated

import httpx
from fastapi import Depends, Request
from pydantic import BaseModel, Field

from app.deps.client_ip import get_client_ip

logger = logging.getLogger(__name__)

BLOCK_THRESHOLD = 80


class RiskProfile(BaseModel):
    ip: str
    threat_score: int = 0
    is_vpn: bool = False
    is_proxy: bool = False
    is_residential_proxy: bool = False
    is_tor: bool = False
    is_anonymous: bool = False
    is_cloud_provider: bool = False
    is_corporate_gateway: bool = False
    vpn_provider_names: list[str] = Field(default_factory=list)
    proxy_provider_names: list[str] = Field(default_factory=list)

    @property
    def should_block(self) -> bool:
        # A corporate gateway scores like a proxy but is a legitimate employer
        # egress. Blocking it locks out everyone behind that company's network.
        if self.is_corporate_gateway:
            return False
        return self.threat_score >= BLOCK_THRESHOLD


async def get_risk(
    request: Request,
    ip: Annotated[str | None, Depends(get_client_ip)],
) -> RiskProfile | None:
    if ip is None:
        return None

    api_key = os.environ.get("IPGEO_API_KEY")
    if not api_key:
        return None

    try:
        response = await request.app.state.http.get(
            "/security", params={"apiKey": api_key, "ip": ip}
        )
        response.raise_for_status()
        security = (response.json() or {}).get("security") or {}
    except (httpx.HTTPError, ValueError) as exc:
        logger.warning("Security lookup failed for %s: %s", ip, exc)
        return None

    return RiskProfile(ip=ip, **{
        key: security[key] for key in RiskProfile.model_fields
        if key in security
    })

That dict comprehension keeps the model and the parser from drifting apart when you add a field. It also means an unexpected key in the response is ignored rather than raising.

Composing them without doubling your calls

Both dependencies declare get_client_ip. FastAPI's dependency injection caches sub-dependency results within a single request by default, so get_client_ip runs once even though two dependencies asked for it. Same applies if a route declares get_risk and a router-level dependency also uses it. You get one call, not two.

If you ever need the opposite, Depends(get_client_ip, use_cache=False) forces a re-run. I've needed that exactly never for this pattern, but it's there.

Wiring it to routes

A type alias per dependency keeps the signatures readable:

# app/routes.py
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException

from app.deps.geo import GeoContext, get_geo
from app.deps.risk import RiskProfile, get_risk

Geo = Annotated[GeoContext | None, Depends(get_geo)]
Risk = Annotated[RiskProfile | None, Depends(get_risk)]

router = APIRouter()


@router.get("/health")
async def health() -> dict[str, str]:
    # No dependencies. No network calls. This is the whole point.
    return {"status": "ok"}


@router.get("/pricing")
async def pricing(geo: Geo) -> dict[str, str]:
    return {"currency": geo.currency_code if geo else "USD"}


@router.post("/signup")
async def signup(geo: Geo, risk: Risk) -> dict[str, str]:
    if risk and risk.should_block:
        logger.info("Blocked signup from %s, score %s", risk.ip, risk.threat_score)
        raise HTTPException(status_code=403, detail="Signup unavailable")
    return {"country": geo.country_code if geo else "unknown"}

Three routes, three different policies, one set of dependencies. /health does nothing, /pricing pays for geo only, /signup takes both. Middleware cannot express that without a path blocklist.

For a whole section of the API, put the check on the router. The catch worth knowing: router-level dependencies run but their return value is thrown away, so they're for enforcement, not injection.

# app/routes.py, continued. Imports and the Geo/Risk aliases are above.


async def require_acceptable_risk(risk: Risk) -> None:
    """Router-level guard. Return value is discarded, so it raises or passes."""
    if risk and risk.should_block:
        raise HTTPException(status_code=403, detail="Unavailable")


checkout = APIRouter(
    prefix="/checkout",
    dependencies=[Depends(require_acceptable_risk)],
)


@checkout.post("/confirm")
async def confirm(risk: Risk) -> dict[str, int]:
    # get_risk already ran for the guard. The cache means this is free.
    return {"score": risk.threat_score if risk else 0}

Testing, which is where this pays off

app.dependency_overrides swaps any dependency for a fake. No HTTP mocking, no responses library, no network in the test suite:

# tests/test_pricing.py
import pytest
from fastapi.testclient import TestClient

from app.deps.geo import GeoContext, get_geo
from app.main import app


def fake_geo_sweden() -> GeoContext:
    return GeoContext(
        ip="91.128.103.196",
        country_code="SE",
        is_eu=True,
        currency_code="SEK",
    )


@pytest.fixture
def client():
    app.dependency_overrides[get_geo] = fake_geo_sweden
    try:
        yield TestClient(app)
    finally:
        # Clear or the override leaks into every later test in the session.
        app.dependency_overrides.clear()


def test_pricing_uses_local_currency(client):
    assert client.get("/pricing").json()["currency"] == "SEK"

Testing the same logic in middleware means intercepting an outbound HTTP call, which is doable but noisier. The testing-dependencies docs cover the override mechanics if you want the full picture.

One HTTP client for the process

Both dependencies reach for request.app.state.http. Create that once in the lifespan handler, not per request, because a fresh AsyncClient throws away connection pooling and TLS session reuse on every call:

# app/main.py
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI

from app.routes import checkout, router


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(
        base_url="https://api.ipgeolocation.io/v3",
        # Separate read timeout because a slow upstream is the realistic
        # failure, not a slow handshake. Keep the total well under your
        # own request budget.
        timeout=httpx.Timeout(connect=1.0, read=1.5, write=1.0, pool=1.0),
    )
    try:
        yield
    finally:
        await app.state.http.aclose()


app = FastAPI(lifespan=lifespan)
app.include_router(router)
app.include_router(checkout)

Caching with Redis

The same IP will hit you many times in a session, and paying for a lookup on every request is wasteful whichever provider you use. Redis over an in-process dict, for one specific reason: with four Uvicorn workers, an in-process cache gives you four separate caches and a hit rate a quarter of what you think you're getting. That bug is invisible in staging with one worker.

Two TTLs, because the data ages differently. Geolocation for an IP is stable for days. Threat signals are not, since an address can be clean this morning and a residential proxy exit by lunchtime.

# app/cache.py
import json
import logging

import redis.asyncio as redis

logger = logging.getLogger(__name__)

GEO_TTL = 24 * 60 * 60   # geolocation barely moves
RISK_TTL = 15 * 60       # threat signals go stale fast


async def cached_json(
    client: redis.Redis, key: str, ttl: int, loader
) -> dict | None:
    """Read-through cache. Redis being down degrades to a direct call."""
    try:
        hit = await client.get(key)
        if hit is not None:
            return json.loads(hit)
    except (redis.RedisError, ValueError) as exc:
        logger.warning("Cache read failed for %s: %s", key, exc)

    value = await loader()
    if value is None:
        return None

    try:
        await client.set(key, json.dumps(value), ex=ttl)
    except redis.RedisError as exc:
        logger.warning("Cache write failed for %s: %s", key, exc)

    return value

Wrap the API call in get_geo with cached_json(redis_client, f"geo:{ip}", GEO_TTL, loader) and you're done. Note that Redis failures are caught and logged rather than raised. A cache outage should slow you down, not take you offline.

Tip: Cache geolocation by /24 rather than by exact IP if your traffic is consumer-heavy. Neighbouring addresses in a residential block almost always resolve to the same city, and your hit rate improves noticeably. Don't do this for risk data, where the whole point is per-address precision.

When middleware actually is right

Dependencies aren't a replacement for middleware, and an article claiming otherwise would be selling you something. Middleware is correct whenever the work genuinely applies to every request, or has to happen outside the route's world.

Request ID generation, access logging, CORS, response header mutation, and compression all belong there, and the middleware docs show the shape. So does the thing this article depends on: Uvicorn's proxy header handling is itself middleware, and it works because rewriting the client IP really does apply to every request equally.

The rule of thumb: if the work applies to every request without exception, use middleware. If it applies to some routes, produces a value the route needs, or you want to fake it in tests, use a dependency. Geolocation and risk enrichment are firmly in the second category.

A few extra notes

IPv6 works without changes. ipaddress.ip_address handles both families and the API takes either, so the code above needs nothing extra. Worth an actual test case, though, because IPv6 is where the assumption that an address is four dot-separated numbers usually surfaces.

Rate limits are an ordinary failure. A 429 raises HTTPStatusError, gets caught by the same except block, and fails open like any other error. If you're hitting limits regularly, the Redis TTL is the lever, not a bigger plan.

Bulk endpoints exist for the offline case. If you're enriching historical logs rather than live requests, the bulk lookup takes up to 50,000 addresses per call, and doing that in a loop of single requests is the slow way to learn it exists.

Drop the two dependency files into an existing app and wire them to one route before you touch the rest. The client IP piece is the part that breaks in production and the part you can verify in about a minute: log request.client.host on a deployed route and confirm it isn't your load balancer. If it is, your --forwarded-allow-ips is wrong and everything downstream has been geolocating your own infrastructure.