IP Geolocation in Python with FastAPI: Detect Country, City & Currency in 15 Lines
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
/24subnet. 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.
httpxtimeout 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