Skip to content

infrastructure.enrichers.location_enricher

src.infrastructure.enrichers.location_enricher

Location enricher implementation for IP geolocation.

Resolves IP addresses to geographic locations using MaxMind GeoIP2.

Implementation
  • Uses GeoLite2-City database for city-level geolocation
  • Fail-open: Returns empty result on any errors
  • Private IPs: Returns empty (no meaningful location)
  • TODO: F7.3 - Automate monthly database updates via background jobs
Reference
  • docs/architecture/session-management-architecture.md
  • MaxMind GeoIP2: https://dev.maxmind.com/geoip/docs/databases/city-and-country

Classes

IPLocationEnricher

Location enricher for IP geolocation using MaxMind GeoIP2.

Uses GeoLite2-City database to resolve IP addresses to geographic locations. Database is updated monthly (manual for now, automated in F7.3).

Implements LocationEnricher protocol (structural typing).

Behavior
  • Fail-open: Returns empty result on errors (never blocks session creation)
  • Private IPs: Always return empty (no meaningful location data)
  • Best-effort: Unknown IPs return empty data
  • Lazy loading: Database reader initialized on first use

Parameters:

Name Type Description Default
db_path str | None

Path to GeoLite2-City.mmdb file. If None, geolocation is disabled.

 None
Source code in src/infrastructure/enrichers/location_enricher.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
class IPLocationEnricher:
    """Location enricher for IP geolocation using MaxMind GeoIP2.

    Uses GeoLite2-City database to resolve IP addresses to geographic locations.
    Database is updated monthly (manual for now, automated in F7.3).

    Implements LocationEnricher protocol (structural typing).

    Behavior:
        - Fail-open: Returns empty result on errors (never blocks session creation)
        - Private IPs: Always return empty (no meaningful location data)
        - Best-effort: Unknown IPs return empty data
        - Lazy loading: Database reader initialized on first use

    Args:
        db_path: Path to GeoLite2-City.mmdb file. If None, geolocation is disabled.
    """

    def __init__(
        self,
        logger: LoggerProtocol,
        db_path: str | None = None,
    ) -> None:
        """Initialize location enricher.

        Args:
            logger: Logger for error/debug messages.
            db_path: Path to GeoIP2 database file. Defaults to settings.geoip_db_path.
        """
        self._logger = logger
        self._db_path = db_path or settings.geoip_db_path
        self._reader: geoip2.database.Reader | None = None

    async def enrich(self, ip_address: str) -> LocationEnrichmentResult:
        """Resolve IP address to geographic location using GeoIP2.

        Args:
            ip_address: Client IP address (IPv4 or IPv6).

        Returns:
            LocationEnrichmentResult with location data.
            Returns empty result for:
            - Private/reserved IPs (no meaningful location)
            - Database not configured or missing
            - IP not found in database
            - Any lookup errors (fail-open)
        """
        if not ip_address:
            return LocationEnrichmentResult()

        try:
            # Check if private/reserved IP (no lookup needed)
            if self._is_private_ip(ip_address):
                return LocationEnrichmentResult()

            # Check if database is configured
            if not self._db_path:
                self._logger.debug(
                    "GeoIP database not configured (geoip_db_path is None)"
                )
                return LocationEnrichmentResult()

            # Lazy load database reader
            if self._reader is None:
                self._init_reader()

            # If reader still None (init failed), return empty
            if self._reader is None:
                return LocationEnrichmentResult()

            # Lookup IP in GeoIP2 database
            response = self._reader.city(ip_address)

            # Extract location data (all fields optional in GeoIP2)
            city = response.city.name if response.city.name else None
            country_code = (
                response.country.iso_code if response.country.iso_code else None
            )
            latitude = (
                response.location.latitude if response.location.latitude else None
            )
            longitude = (
                response.location.longitude if response.location.longitude else None
            )

            # Format location string: "City, CC" or "CC" if no city
            location = None
            if city and country_code:
                location = f"{city}, {country_code}"
            elif country_code:
                location = country_code

            return LocationEnrichmentResult(
                location=location,
                city=city,
                country_code=country_code,
                latitude=latitude,
                longitude=longitude,
            )

        except geoip2.errors.AddressNotFoundError:
            # IP not in database (common for some IP ranges)
            self._logger.debug(
                "IP not found in GeoIP database",
                ip_address=ip_address,
            )
            return LocationEnrichmentResult()

        except Exception as e:
            # Any other error: fail-open (log warning, return empty)
            self._logger.warning(
                "Failed to enrich IP location",
                ip_address=ip_address,
                error=str(e),
            )
            return LocationEnrichmentResult()

    def _init_reader(self) -> None:
        """Initialize GeoIP2 database reader (lazy loading).

        Called on first use. If initialization fails, logs warning and sets
        reader to None (geolocation disabled).
        """
        try:
            if not self._db_path:
                return

            db_file = Path(self._db_path)
            if not db_file.exists():
                self._logger.warning(
                    "GeoIP database file not found",
                    db_path=self._db_path,
                )
                return

            self._reader = geoip2.database.Reader(str(db_file))
            self._logger.info(
                "GeoIP database loaded successfully",
                db_path=self._db_path,
            )

        except Exception as e:
            self._logger.warning(
                "Failed to initialize GeoIP database",
                db_path=self._db_path,
                error=str(e),
            )
            self._reader = None

    def _is_private_ip(self, ip_address: str) -> bool:
        """Check if IP address is private/reserved.

        Private IPs have no meaningful geographic location.

        Args:
            ip_address: IP address string.

        Returns:
            True if private/reserved, False if public.
        """
        try:
            ip = ipaddress.ip_address(ip_address)
            return (
                ip.is_private
                or ip.is_loopback
                or ip.is_reserved
                or ip.is_link_local
                or ip.is_multicast
            )
        except ValueError:
            # Invalid IP address format
            return True
Functions
__init__ 
__init__(
    logger: LoggerProtocol, db_path: str | None = None
) -> None

Parameters:

Name Type Description Default
logger LoggerProtocol

Logger for error/debug messages.

 required
db_path str | None

Path to GeoIP2 database file. Defaults to settings.geoip_db_path.

 None
Source code in src/infrastructure/enrichers/location_enricher.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(
    self,
    logger: LoggerProtocol,
    db_path: str | None = None,
) -> None:
    """Initialize location enricher.

    Args:
        logger: Logger for error/debug messages.
        db_path: Path to GeoIP2 database file. Defaults to settings.geoip_db_path.
    """
    self._logger = logger
    self._db_path = db_path or settings.geoip_db_path
    self._reader: geoip2.database.Reader | None = None
enrich async 
enrich(ip_address: str) -> LocationEnrichmentResult

Resolve IP address to geographic location using GeoIP2.

Parameters:

Name Type Description Default
ip_address str

Client IP address (IPv4 or IPv6).

 required

Returns:

Type Description
LocationEnrichmentResult

LocationEnrichmentResult with location data.
LocationEnrichmentResult

Returns empty result for:
LocationEnrichmentResult
  • Private/reserved IPs (no meaningful location)
LocationEnrichmentResult
  • Database not configured or missing
LocationEnrichmentResult
  • IP not found in database
LocationEnrichmentResult
  • Any lookup errors (fail-open)
Source code in src/infrastructure/enrichers/location_enricher.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
async def enrich(self, ip_address: str) -> LocationEnrichmentResult:
    """Resolve IP address to geographic location using GeoIP2.

    Args:
        ip_address: Client IP address (IPv4 or IPv6).

    Returns:
        LocationEnrichmentResult with location data.
        Returns empty result for:
        - Private/reserved IPs (no meaningful location)
        - Database not configured or missing
        - IP not found in database
        - Any lookup errors (fail-open)
    """
    if not ip_address:
        return LocationEnrichmentResult()

    try:
        # Check if private/reserved IP (no lookup needed)
        if self._is_private_ip(ip_address):
            return LocationEnrichmentResult()

        # Check if database is configured
        if not self._db_path:
            self._logger.debug(
                "GeoIP database not configured (geoip_db_path is None)"
            )
            return LocationEnrichmentResult()

        # Lazy load database reader
        if self._reader is None:
            self._init_reader()

        # If reader still None (init failed), return empty
        if self._reader is None:
            return LocationEnrichmentResult()

        # Lookup IP in GeoIP2 database
        response = self._reader.city(ip_address)

        # Extract location data (all fields optional in GeoIP2)
        city = response.city.name if response.city.name else None
        country_code = (
            response.country.iso_code if response.country.iso_code else None
        )
        latitude = (
            response.location.latitude if response.location.latitude else None
        )
        longitude = (
            response.location.longitude if response.location.longitude else None
        )

        # Format location string: "City, CC" or "CC" if no city
        location = None
        if city and country_code:
            location = f"{city}, {country_code}"
        elif country_code:
            location = country_code

        return LocationEnrichmentResult(
            location=location,
            city=city,
            country_code=country_code,
            latitude=latitude,
            longitude=longitude,
        )

    except geoip2.errors.AddressNotFoundError:
        # IP not in database (common for some IP ranges)
        self._logger.debug(
            "IP not found in GeoIP database",
            ip_address=ip_address,
        )
        return LocationEnrichmentResult()

    except Exception as e:
        # Any other error: fail-open (log warning, return empty)
        self._logger.warning(
            "Failed to enrich IP location",
            ip_address=ip_address,
            error=str(e),
        )
        return LocationEnrichmentResult()