Implementing IP Geolocation in Python: From requests to FastAPI

Python is everywhere in backend development: Django apps, FastAPI services, data pipelines, ML inference servers, automation scripts. Each of these is a place IP geolocation can be useful — and each one has slightly different patterns for adding it.

This post covers practical IP geolocation in Python: starting with requests, moving to the official SDK, then middleware patterns for Django and FastAPI, async clients, caching, and the gotchas you’ll only learn by deploying.

We’ll use the Ip2Geo API throughout, but the patterns transfer to any IP geolocation API with a REST interface.

Minimum Viable Version

The simplest way to geolocate an IP in Python:

import os
import requests

resp = requests.get(
    'https://api.ip2geo.dev/convert',
    params={'ip': '8.8.8.8'},
    headers={'X-Api-Key': os.environ['IP2GEO_API_KEY']}
)
data = resp.json()['data']
print(data['continent']['country']['name'])  # "United States"

Three lines, one dependency (requests, which most Python projects already have). For a quick script this is enough.

Using the Official SDK

pip install ip2geo-sdk

import os
import ip2geo

ip2geo.init(os.environ['IP2GEO_API_KEY'])

result = ip2geo.convert_ip(ip='8.8.8.8')
if result['success']:
    country = result['data']['continent']['country']['name']
    print(country)

The SDK handles authentication, retries, and (optionally) in-process caching for you. For production Python code, this is the cleaner path.

For batch lookups:

result = ip2geo.convert_ips(ips=['8.8.8.8', '1.1.1.1', '9.9.9.9'])

for entry in result['data']:
    print(entry['ip'], entry['continent']['country']['name'])

Getting the Real Client IP

In Python web frameworks, the framework will typically expose the client IP — but you have to use the right field. The naive choices break in production behind a load balancer or reverse proxy.

Django

def get_client_ip(request):
    xff = request.META.get('HTTP_X_FORWARDED_FOR')
    if xff:
        return xff.split(',')[0].strip()
    cf = request.META.get('HTTP_CF_CONNECTING_IP')
    if cf:
        return cf
    return request.META.get('REMOTE_ADDR', '')

Django by default trusts REMOTE_ADDR (the socket address). Behind a proxy that’s the proxy’s IP, not the user’s. Use X-Forwarded-For instead — but only when you control the proxy that sets it.

FastAPI / Starlette

from fastapi import Request

def get_client_ip(request: Request) -> str:
    xff = request.headers.get('x-forwarded-for')
    if xff:
        return xff.split(',')[0].strip()
    cf = request.headers.get('cf-connecting-ip')
    if cf:
        return cf
    return request.client.host if request.client else ''

Flask

from flask import request

def get_client_ip():
    if 'X-Forwarded-For' in request.headers:
        return request.headers['X-Forwarded-For'].split(',')[0].strip()
    return request.remote_addr or ''

Or use Werkzeug’s ProxyFix middleware, which adjusts request.remote_addr based on a configured number of trusted proxies.

Critical reminder: only trust forwarded headers when your proxy controls them. Otherwise users can spoof their IP by setting the header themselves.

Django Middleware Pattern

import ip2geo
from django.conf import settings

ip2geo.init(settings.IP2GEO_API_KEY)

class GeolocationMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        ip = self._get_client_ip(request)
        request.geo = None
        if ip and not ip.startswith(('127.', '10.', '192.168.', '172.')):
            try:
                result = ip2geo.convert_ip(ip=ip)
                if result['success']:
                    request.geo = result['data']
            except Exception as e:
                # Don't fail the request over an enrichment failure
                import logging
                logging.getLogger(__name__).warning('geo lookup failed: %s', e)
        return self.get_response(request)
    
    @staticmethod
    def _get_client_ip(request):
        xff = request.META.get('HTTP_X_FORWARDED_FOR')
        if xff:
            return xff.split(',')[0].strip()
        return request.META.get('REMOTE_ADDR', '')

Add to MIDDLEWARE in settings:

MIDDLEWARE = [
    # ... other middleware ...
    'myapp.middleware.GeolocationMiddleware',
]

Now any view has request.geo available:

def home(request):
    country = (
        request.geo['continent']['country']['name']
        if request.geo else 'Unknown'
    )
    return HttpResponse(f'Hello from {country}')

FastAPI Middleware Pattern

import ip2geo
from fastapi import FastAPI, Request
import os

ip2geo.init(os.environ['IP2GEO_API_KEY'])
app = FastAPI()

@app.middleware('http')
async def geo_middleware(request: Request, call_next):
    ip = request.headers.get('x-forwarded-for', '').split(',')[0].strip()
    if not ip:
        ip = request.client.host if request.client else ''
    
    request.state.geo = None
    if ip and not ip.startswith(('127.', '10.', '192.168.', '172.')):
        try:
            result = ip2geo.convert_ip(ip=ip)
            if result['success']:
                request.state.geo = result['data']
        except Exception:
            pass
    
    return await call_next(request)

@app.get('/')
async def root(request: Request):
    country = (
        request.state.geo['continent']['country']['name']
        if request.state.geo else 'Unknown'
    )
    return {'message': f'Hello from {country}'}

The middleware runs on every request, populates request.state.geo, and your handlers branch on it.

Async Clients

The SDK calls (and requests) are synchronous. If you’re in an async framework like FastAPI, calling them blocks the event loop — bad. Either:

1. Run the sync SDK in a thread

import asyncio
import ip2geo

result = await asyncio.to_thread(ip2geo.convert_ip, ip='8.8.8.8')

Simple, works, but uses a thread per call.

2. Use httpx.AsyncClient directly

import httpx

async def lookup_ip(ip: str, api_key: str):
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            'https://api.ip2geo.dev/convert',
            params={'ip': ip},
            headers={'X-Api-Key': api_key}
        )
        return resp.json()

@app.get('/lookup/{ip}')
async def lookup(ip: str):
    data = await lookup_ip(ip, os.environ['IP2GEO_API_KEY'])
    return data

True async, no thread pool overhead. For high-QPS services, this is the right choice.

Caching: Don’t Pay Twice

The middleware above calls the API on every request. With real traffic, you want to cache.

In-process LRU (single instance)

from cachetools import TTLCache
import ip2geo

geo_cache = TTLCache(maxsize=10_000, ttl=60)

def lookup_with_cache(ip: str):
    if ip in geo_cache:
        return geo_cache[ip]
    try:
        result = ip2geo.convert_ip(ip=ip)
        if result['success']:
            geo_cache[ip] = result['data']
            return result['data']
    except Exception:
        return None
    return None

cachetools is the standard library for in-process caching with TTL. Simple, fast, no external dependencies beyond the library itself.

Distributed cache with Redis

For multi-instance services where the same IP might land on different servers:

import json
import redis

r = redis.Redis(host='localhost', port=6379)

def lookup_with_redis_cache(ip: str):
    cache_key = f'geo:{ip}'
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)
    
    result = ip2geo.convert_ip(ip=ip)
    if result['success']:
        r.setex(cache_key, 300, json.dumps(result['data']))
        return result['data']
    return None

5-minute TTL is a reasonable starting point. See caching strategies for tuning advice.

Bulk Lookups for Pipelines

For data pipelines processing many IPs:

import ip2geo

# Process 50 IPs in one HTTP round-trip
ips = [...your batch of IPs...]
result = ip2geo.convert_ips(ips=ips)

for entry in result['data']:
    process(entry['ip'], entry['continent']['country']['code'])

For pandas dataframes:

import pandas as pd
import ip2geo

df = pd.read_csv('access_logs.csv')

# Get unique IPs to minimize API calls
unique_ips = df['ip'].unique().tolist()

# Batch lookup
result = ip2geo.convert_ips(ips=unique_ips)

# Build lookup map
geo_by_ip = {
    entry['ip']: entry['continent']['country']['code']
    for entry in result['data']
}

# Apply to dataframe
df['country'] = df['ip'].map(geo_by_ip)

Common Pitfalls

1. Blocking the event loop

In async frameworks (FastAPI, Starlette, asyncio code), don’t use sync HTTP libraries directly. Use httpx.AsyncClient or wrap sync calls in asyncio.to_thread.

2. Forgetting trust_proxy

Behind any reverse proxy, the IP your framework reports by default is the proxy’s IP, not the user’s. Always configure your framework to read forwarded headers.

3. Logging IPs without retention limits

GDPR treats IPs as personal data. Default to logging country/ASN; only log raw IPs with a defined retention.

4. No fallback for geo failures

The API will occasionally be slow or fail. Your code should treat geo lookup as best-effort enrichment, not a hard dependency.

5. Calling per-row in a tight loop

If you’re processing 1 million rows in a pandas dataframe, don’t make 1 million API calls. Get the unique IPs first, batch-lookup, then map back.

6. Hardcoding the API key

Read from environment variables. Use a .env file for local dev, real secrets management for production.

Pure-Python Patterns Without Async

For Flask, sync Django, scripts, and CLI tools, the synchronous pattern is fine:

# Sync pattern — simple and works
import ip2geo
ip2geo.init(os.environ['IP2GEO_API_KEY'])

def enrich(ip):
    result = ip2geo.convert_ip(ip=ip)
    return result['data'] if result['success'] else None

Most Python web apps don’t need async to handle their actual load. Don’t over-engineer. Use async when it solves an actual problem.

A Production-Ready FastAPI Skeleton

import os
from cachetools import TTLCache
import ip2geo
from fastapi import FastAPI, Request

ip2geo.init(os.environ['IP2GEO_API_KEY'])

app = FastAPI()
geo_cache = TTLCache(maxsize=10_000, ttl=300)

def get_client_ip(request: Request) -> str:
    xff = request.headers.get('x-forwarded-for', '')
    if xff:
        return xff.split(',')[0].strip()
    return request.client.host if request.client else ''

@app.middleware('http')
async def geo_middleware(request: Request, call_next):
    ip = get_client_ip(request)
    request.state.geo = None
    
    if ip and not ip.startswith(('127.', '10.', '192.168.', '::1')):
        if ip in geo_cache:
            request.state.geo = geo_cache[ip]
        else:
            try:
                # Run sync SDK call in a thread to avoid blocking
                import asyncio
                result = await asyncio.to_thread(ip2geo.convert_ip, ip=ip)
                if result['success']:
                    request.state.geo = result['data']
                    geo_cache[ip] = result['data']
            except Exception as e:
                import logging
                logging.warning(f'geo lookup failed: {e}')
    
    return await call_next(request)

@app.get('/')
async def root(request: Request):
    geo = request.state.geo
    return {
        'country': geo['continent']['country']['code'] if geo else None,
        'city': geo['continent']['country']['city']['name'] if geo else None,
    }

Drop-in, cache-backed, non-blocking, fail-soft. Adjust cache size and TTL for your traffic shape.

TL;DR

• Use requests for quick scripts; use the SDK for production.
• Get the client IP from forwarded headers, not REMOTE_ADDR, and only with a trusted proxy in front.
• Add middleware once — every handler then has request.geo or request.state.geo available.
• Cache aggressively — cachetools.TTLCache for single-instance, Redis for distributed.
• In async code, use httpx.AsyncClient or wrap sync calls in asyncio.to_thread.
• Batch bulk lookups with convert_ips([...]).
• Don’t expose your API key in client-side code, in repo, or in logs.

If your stack also includes Node.js services, the same patterns apply — see the Node.js implementation guide. Both SDKs call the same API and return identical data structures, so cross-language integrations stay consistent.