PythonFastAPITutorialGeolocationExchange Rates

IP Geolocation in Python with FastAPI: Detect Country, City & Currency in 15 Lines

ApogeoAPI6 min read

Most geo-detection tutorials for Python show a raw requests.get() call to an IP API. That works in a script, but in production you need:

  • The real client IP behind Nginx / a load balancer
  • Caching (calling an external API on every request eats your quota in hours)
  • Graceful fallback when the API is down or the IP is a private address

This tutorial covers all three, using FastAPI as the primary framework with a Django note at the end.

Install

pip install fastapi httpx uvicorn redis

We use httpx (async HTTP) instead of requests so we don't block FastAPI's event loop.

A geo module that handles caching

# geo.py
import os
import hashlib
import httpx
import redis.asyncio as aioredis

APOGEO_KEY = os.environ["APOGEOAPI_KEY"]
APOGEO_BASE = "https://api.apogeoapi.com/v1"
CACHE_TTL = 3600  # 1 hour per subnet

_redis: aioredis.Redis | None = None

def get_redis() -> aioredis.Redis:
    global _redis
    if _redis is None:
        _redis = aioredis.from_url(os.environ.get("REDIS_URL", "redis://localhost:6379"))
    return _redis


def subnet_key(ip: str) -> str:
    """Cache by /24 subnet — everyone in the same office gets the same result."""
    parts = ip.split(".")
    if len(parts) == 4:
        subnet = ".".join(parts[:3]) + ".0"
    else:
        subnet = ip  # IPv6: cache by full address
    return "geo:" + hashlib.md5(subnet.encode()).hexdigest()


async def lookup_ip(ip: str) -> dict:
    """Return geo data for an IP. Cached per /24 subnet for CACHE_TTL seconds."""
    cache = get_redis()
    key = subnet_key(ip)

    cached = await cache.get(key)
    if cached:
        import json
        return json.loads(cached)

    try:
        async with httpx.AsyncClient(timeout=3.0) as client:
            r = await client.get(
                f"{APOGEO_BASE}/ip/{ip}",
                headers={"X-API-Key": APOGEO_KEY},
            )
            r.raise_for_status()
            data = r.json()
    except Exception:
        return {}  # graceful fallback

    import json
    await cache.setex(key, CACHE_TTL, json.dumps(data))
    return data

Key design choices:

  • Subnet caching: Two devices at the same office share a /24 subnet. Caching by subnet rather than exact IP reduces API calls by 10–50x without meaningfully sacrificing accuracy for country/city detection.
  • 3-second timeout: IP lookups should never block a request for more than a second. httpx timeout ensures the request completes fast or fails fast.
  • Empty dict fallback: If the API is unreachable or the IP is a local address (127.0.0.1, 10.x.x.x), the endpoint still returns — it just won't have geo data.

FastAPI — extract the real IP

# main.py
from fastapi import FastAPI, Request
from geo import lookup_ip

app = FastAPI()


def get_client_ip(request: Request) -> str:
    """
    Real IP behind Nginx / Cloudflare / AWS ALB.
    Never trust X-Forwarded-For from untrusted clients directly —
    but it's safe to read the *first* entry when you control your proxy.
    """
    # Cloudflare sets this header reliably
    if cf_ip := request.headers.get("cf-connecting-ip"):
        return cf_ip
    # Standard proxy header — take the leftmost (original client)
    if xff := request.headers.get("x-forwarded-for"):
        return xff.split(",")[0].strip()
    # Direct connection (dev / no proxy)
    return request.client.host or "127.0.0.1"


@app.get("/api/geo")
async def geo_endpoint(request: Request):
    ip = get_client_ip(request)
    data = await lookup_ip(ip)

    country = data.get("country", {})
    city = data.get("city", {})

    return {
        "ip": ip,
        "country": country.get("name"),
        "country_code": country.get("iso2"),
        "city": city.get("name"),
        "currency": country.get("currency"),
        "currency_rate": country.get("currencyRate"),  # USD → local
        "timezone": data.get("timezone"),
        "flag_url": country.get("flagUrl"),
    }

Run it:

APOGEOAPI_KEY=your_key uvicorn main:app --reload
curl http://localhost:8000/api/geo
# {
#   "ip": "190.210.x.x",
#   "country": "Argentina",
#   "country_code": "AR",
#   "city": "Buenos Aires",
#   "currency": "ARS",
#   "currency_rate": 1048.5,
#   "timezone": "America/Argentina/Buenos_Aires",
#   "flag_url": "https://flagcdn.com/ar.svg"
# }

Middleware pattern — geo data available on every request

For apps where many routes need the visitor's country (pricing, localization, analytics), it's cleaner to resolve it once in middleware and store it on request.state:

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from geo import lookup_ip, get_client_ip  # move get_client_ip to geo.py

app = FastAPI()

class GeoMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        ip = get_client_ip(request)
        request.state.geo = await lookup_ip(ip)
        response = await call_next(request)
        return response

app.add_middleware(GeoMiddleware)


@app.get("/pricing")
async def pricing(request: Request):
    geo = request.state.geo
    country = geo.get("country", {})
    usd_price = 19.0
    rate = country.get("currencyRate", 1.0)
    local_price = round(usd_price * rate, 2)
    currency = country.get("currency", "USD")
    return {"price_usd": usd_price, "price_local": local_price, "currency": currency}

Django equivalent

In Django, add a middleware class in geo_middleware.py:

import httpx
from django.core.cache import cache

APOGEO_KEY = "your_key"

class GeoMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        ip = (
            request.META.get("HTTP_CF_CONNECTING_IP")
            or (request.META.get("HTTP_X_FORWARDED_FOR") or "").split(",")[0].strip()
            or request.META.get("REMOTE_ADDR", "127.0.0.1")
        )
        cache_key = f"geo:{ip}"
        geo = cache.get(cache_key)
        if geo is None:
            try:
                r = httpx.get(
                    f"https://api.apogeoapi.com/v1/ip/{ip}",
                    headers={"X-API-Key": APOGEO_KEY},
                    timeout=3,
                )
                geo = r.json() if r.is_success else {}
            except Exception:
                geo = {}
            cache.set(cache_key, geo, timeout=3600)
        request.geo = geo
        return self.get_response(request)

Register it in settings.py:

MIDDLEWARE = [
    # ... existing middleware ...
    "myapp.geo_middleware.GeoMiddleware",
]

Then in any view: request.geo.get("country", {}).get("name")

Getting the exchange rate directly

If you only need an exchange rate (not IP lookup), call the rate endpoint directly:

async def get_rate(currency: str) -> float:
    """Return USD → currency rate. Cached 4 hours."""
    cache = get_redis()
    key = f"rate:{currency.upper()}"
    cached = await cache.get(key)
    if cached:
        return float(cached)

    async with httpx.AsyncClient(timeout=5) as client:
        r = await client.get(
            f"https://api.apogeoapi.com/v1/exchange-rates/{currency.upper()}",
            headers={"X-API-Key": APOGEO_KEY},
        )
        r.raise_for_status()
        rate = r.json()["usdRate"]

    await cache.setex(key, 14400, str(rate))  # 4h TTL
    return rate

Quota math

With subnet caching at 1-hour TTL:

  • 1,000 daily unique /24 subnets → ~720 cached hits per subnet per day → 1,000 API calls covers ~720K requests
  • The free tier (1,000 req/month) handles roughly a 3K daily-active-user app with zero cost
  • The Basic tier (15K req/month) handles ~45K DAU

Get a free key at apogeoapi.com — no credit card, 1-minute signup.

Try ApogeoAPI free

1,000 requests/month forever. 14-day full-access trial. No credit card.

Get your free API key