IP Geolocation in Python: Official SDK, REST Calls and Framework Recipes

pip install iplookupapi
# or
uv add iplookupapi
# or
poetry add iplookupapi

export IPLOOKUPAPI_KEY="your-api-key"

import os

import iplookupapi

client = iplookupapi.Client(os.environ["IPLOOKUPAPI_KEY"])

result = client.info("8.8.8.8")  # a dict parsed from the JSON response
data = result["data"]

print(data["location"]["country"]["name"])
print(data["location"]["city"]["name"])
print(data["connection"]["asn"], data["connection"]["isp"])
print(data["timezone"]["id"])

import os

import iplookupapi

client = iplookupapi.Client(os.environ["IPLOOKUPAPI_KEY"])

own = client.info()  # no IP: the API answers for the address the request came from
german = client.info("8.8.8.8", "de")  # positional arguments: ip, language
status = client.status()  # free, does not count against your quota

print(own["data"]["ip"])
print(german["data"]["location"]["country"]["name_translated"])
print(status["quotas"]["month"]["remaining"])

import os

import requests

resp = requests.get(
    "https://api.iplookupapi.com/v1/info",
    params={"ip": "8.8.8.8", "fields": "ip,location.country.alpha2,timezone.id"},
    headers={"apikey": os.environ["IPLOOKUPAPI_KEY"]},
    timeout=10,
)
resp.raise_for_status()
print(resp.json()["data"])

import ipaddress
import os
import random
import time

import requests

API_BASE = "https://api.iplookupapi.com/v1"

session = requests.Session()
session.headers["apikey"] = os.environ["IPLOOKUPAPI_KEY"]


class IPLookupError(Exception):
    def __init__(self, status, body):
        self.status = status
        self.body = body
        self.code = (body.get("error") or {}).get("code")
        super().__init__(f"HTTP {status}: {body.get('message')}")


def _call(method, path, retries=4, **kwargs):
    kwargs.setdefault("timeout", (3.05, 15))  # (connect, read) in seconds
    for attempt in range(retries + 1):
        try:
            resp = session.request(method, API_BASE + path, **kwargs)
        except (requests.ConnectionError, requests.Timeout):
            if attempt == retries:
                raise
        else:
            if resp.status_code == 200:
                return resp.json()
            try:
                body = resp.json()
            except ValueError:
                body = {"message": resp.text}
            code = (body.get("error") or {}).get("code")
            # Retry server errors and the per-minute limit, never a used-up quota.
            retry = resp.status_code >= 500 or (
                resp.status_code == 429
                and code != "quota_exceeded"
                and "quota" not in body
            )
            if not retry or attempt == retries:
                raise IPLookupError(resp.status_code, body)
        time.sleep(min(30, 2 ** attempt) + random.random())  # backoff + jitter


def lookup(ip=None, fields=None, language=None, hostname=False):
    """One address (or the caller's own IP if ip is None). Returns the data object."""
    params = {"ip": ip, "fields": fields, "language": language}
    if hostname:
        params["hostname"] = "true"
    return _call("GET", "/info", params=params)["data"]


def lookup_many(ips, fields=None, language=None, chunk_size=1000):
    """Any number of addresses, sent in POST batches of up to 1,000."""
    valid, results, errors = [], {}, {}
    for ip in dict.fromkeys(ips):  # dedupe, keep order
        try:
            ipaddress.ip_address(ip)
            valid.append(ip)
        except ValueError:  # a chunk without any valid IP would fail with 422
            errors[ip] = "Not a valid IP address."
    for start in range(0, len(valid), chunk_size):
        body = _call(
            "POST", "/batch",
            params={"fields": fields, "language": language},
            json={"ips": valid[start:start + chunk_size]},
            timeout=(3.05, 60),
        )
        results.update(body.get("data") or {})
        errors.update(body.get("errors") or {})
    return results, errors


def remaining_quota():
    """GET /v1/status is free and does not count against the quota."""
    return _call("GET", "/status")["quotas"]["month"]["remaining"]


def dig(obj, path):
    """Safe dot-path access: dig(data, "location.country.alpha2")."""
    for key in path.split("."):
        if not isinstance(obj, dict):
            return None
        obj = obj.get(key)
    return obj

import os

import iplookupapi
from everapi.exceptions import (
    ApiError, IncorrectApikey, NotAllowed, QuotaExceeded, RateLimitExceeded,
)

client = iplookupapi.Client(os.environ["IPLOOKUPAPI_KEY"])


def country_code(ip):
    try:
        result = client.info(ip)
    except IncorrectApikey:
        raise SystemExit("Check IPLOOKUPAPI_KEY (HTTP 401).")
    except NotAllowed:
        raise SystemExit("Your plan or key does not allow this request (HTTP 403).")
    except QuotaExceeded:
        raise SystemExit("Monthly quota used up (HTTP 429).")
    except RateLimitExceeded:
        return None  # per-minute limit: retry later
    except ApiError as exc:  # HTTP 422: exc.args[1] holds the validation errors
        print("Invalid request:", exc.args[1])
        return None
    if "data" not in result:  # for example HTTP 404 "No data found for this IP"
        return None
    return result["data"]["location"]["country"]["alpha2"]

from flask import Flask, jsonify, request
from werkzeug.middleware.proxy_fix import ProxyFix

from iplookup import IPLookupError, lookup

app = Flask(__name__)
# Exactly one reverse proxy (for example nginx) sets X-Forwarded-For.
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1)

FIELDS = "ip,location.country.alpha2,location.city.name,timezone.id"


@app.get("/geo")
def geo():
    try:
        data = lookup(request.remote_addr, fields=FIELDS)
    except IPLookupError as exc:
        return jsonify(error=str(exc)), 502
    return jsonify(data)

import ipaddress

from django.conf import settings


def _parse(value):
    try:
        ip = ipaddress.ip_address(value.strip())
    except ValueError:
        return None
    return getattr(ip, "ipv4_mapped", None) or ip  # ::ffff:203.0.113.7 -> 203.0.113.7


def client_ip(request):
    """REMOTE_ADDR, or the right-most X-Forwarded-For entry that is not a trusted proxy."""
    trusted = [ipaddress.ip_network(n) for n in getattr(settings, "TRUSTED_PROXIES", [])]
    ip = _parse(request.META.get("REMOTE_ADDR", ""))
    hops = [h for h in request.META.get("HTTP_X_FORWARDED_FOR", "").split(",") if h.strip()]
    while ip is not None and hops and any(ip in net for net in trusted):
        ip = _parse(hops.pop())
    return str(ip) if ip is not None else None

from django.http import JsonResponse

from iplookup import IPLookupError, lookup
from .utils import client_ip


def geo(request):
    ip = client_ip(request)
    if ip is None:
        return JsonResponse({"error": "No valid client IP"}, status=400)
    try:
        data = lookup(ip, fields="ip,location.country.alpha2,timezone.id")
    except IPLookupError as exc:
        return JsonResponse({"error": str(exc)}, status=502)
    return JsonResponse(data)

import os
from contextlib import asynccontextmanager

import httpx
from fastapi import Depends, FastAPI, HTTPException, Request

FIELDS = "ip,location.country.alpha2,location.city.name,timezone.id"


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(
        base_url="https://api.iplookupapi.com/v1",
        headers={"apikey": os.environ["IPLOOKUPAPI_KEY"]},
        timeout=httpx.Timeout(10.0, connect=3.0),
    )
    yield
    await app.state.http.aclose()


app = FastAPI(lifespan=lifespan)


async def visitor_geo(request: Request) -> dict:
    params = {"fields": FIELDS}
    if request.client:
        params["ip"] = request.client.host
    try:
        resp = await request.app.state.http.get("/info", params=params)
    except httpx.HTTPError:
        raise HTTPException(status_code=503, detail="IP lookup unavailable")
    if resp.status_code != 200:
        raise HTTPException(status_code=502, detail=f"IP lookup failed: HTTP {resp.status_code}")
    return resp.json()["data"]


@app.get("/geo")
async def geo(geo: dict = Depends(visitor_geo)):
    location = geo.get("location") or {}  # private and loopback addresses have no location
    return {"country": (location.get("country") or {}).get("alpha2"), "timezone": geo.get("timezone")}

import pandas as pd

from iplookup import dig, lookup_many

df = pd.read_csv("visits.csv", dtype={"ip": "string"})  # needs a column named "ip"
df["ip"] = df["ip"].str.strip()
fields = "ip,location.country.alpha2,location.city.name,connection.asn,connection.isp"

results, errors = lookup_many(df["ip"].dropna(), fields=fields)

columns = {
    "country": "location.country.alpha2",
    "city": "location.city.name",
    "asn": "connection.asn",
    "isp": "connection.isp",
}
for column, path in columns.items():
    df[column] = df["ip"].map(lambda ip: dig(results.get(ip), path))
df["asn"] = df["asn"].astype("Int64")

df.to_csv("visits_enriched.csv", index=False)
print(f"{len(results)} addresses enriched, {len(errors)} not valid IPs")

from collections import Counter

from iplookup import dig, lookup_many

with open("access.log", encoding="utf-8") as f:
    ips = [line.split(" ", 1)[0] for line in f if line.strip()]

results, errors = lookup_many(ips, fields="ip,location.country.alpha2")
countries = Counter(dig(results.get(ip), "location.country.alpha2") or "unknown" for ip in ips)
print(countries.most_common(10))

import time
from functools import lru_cache

from iplookup import dig, lookup


@lru_cache(maxsize=10_000)
def country_of(ip):
    return dig(lookup(ip, fields="ip,location.country.alpha2"), "location.country.alpha2")


TTL_SECONDS = 6 * 3600
_cache = {}


def cached_lookup(ip, fields=None):
    key = (ip, fields)
    hit = _cache.get(key)
    if hit and hit[0] > time.monotonic():
        return hit[1]
    data = lookup(ip, fields=fields)
    _cache[key] = (time.monotonic() + TTL_SECONDS, data)
    return data