"""Geo service — backward compatibility wrappers. DEPRECATED: This module is kept for backward compatibility only. New code should use :class:`GeoCache` directly from ``app.services.geo_cache``. The underlying implementation has been refactored to eliminate module-level mutable state. The :class:`GeoCache` instance should be injected into services and routers via dependency injection, and stored on ``app.state.geo_cache``. See :class:`app.services.geo_cache.GeoCache` for the implementation. """ from __future__ import annotations from typing import TYPE_CHECKING from app.services.geo_cache import GeoCache from app.models.geo import GeoInfo if TYPE_CHECKING: import aiohttp import aiosqlite __all__ = [ "GeoCache", "clear_cache", "clear_neg_cache", "is_cached", "lookup", "lookup_batch", "lookup_cached_only", "cache_stats", "count_unresolved", "get_unresolved_ips", "load_cache_from_db", "flush_dirty", "re_resolve_all", "init_geoip", ] #: Deprecated: Module-level cache instance for backward compatibility. #: This exists only to provide backward-compatible module-level functions. #: New code should inject GeoCache instances directly. _default_geo_cache: GeoCache = GeoCache() async def clear_cache() -> None: """(DEPRECATED) Flush both the positive and negative lookup caches. Use :meth:`GeoCache.clear` instead. This function delegates to the default module-level instance for backward compatibility only. """ await _default_geo_cache.clear() async def clear_neg_cache() -> None: """(DEPRECATED) Flush only the negative (failed-lookups) cache. Use :meth:`GeoCache.clear_neg_cache` instead. This function delegates to the default module-level instance for backward compatibility only. """ await _default_geo_cache.clear_neg_cache() def is_cached(ip: str) -> bool: """(DEPRECATED) Return ``True`` if *ip* has a positive cache entry. Use :meth:`GeoCache.is_cached` instead. This function delegates to the default module-level instance for backward compatibility only. Args: ip: IPv4 or IPv6 address string. Returns: ``True`` when *ip* is in the cache with a known country code. """ return _default_geo_cache.is_cached(ip) async def cache_stats(db: aiosqlite.Connection) -> dict[str, int]: """(DEPRECATED) Return diagnostic counters for the geo cache subsystem. Use :meth:`GeoCache.cache_stats` instead. This function delegates to the default module-level instance for backward compatibility only. Args: db: Open BanGUI application database connection. Returns: Dict with keys ``cache_size``, ``unresolved``, ``neg_cache_size``, and ``dirty_size``. """ return await _default_geo_cache.cache_stats(db) async def count_unresolved(db: aiosqlite.Connection) -> int: """(DEPRECATED) Return the number of unresolved entries in the geo cache. Use :meth:`GeoCache.count_unresolved` instead. """ return await _default_geo_cache.count_unresolved(db) async def get_unresolved_ips(db: aiosqlite.Connection) -> list[str]: """(DEPRECATED) Return IPs with NULL country_code in the persistent cache. Use :meth:`GeoCache.get_unresolved_ips` instead. Args: db: Open BanGUI application database connection. Returns: List of IP addresses that are candidates for re-resolution. """ return await _default_geo_cache.get_unresolved_ips(db) def init_geoip(mmdb_path: str | None) -> None: """(DEPRECATED) Initialise the MaxMind GeoLite2-Country database reader. Use :meth:`GeoCache.init_geoip` instead. This function delegates to the default module-level instance for backward compatibility only. Args: mmdb_path: Absolute path to a ``GeoLite2-Country.mmdb`` file. """ _default_geo_cache.init_geoip(mmdb_path) async def load_cache_from_db(db: aiosqlite.Connection) -> None: """(DEPRECATED) Pre-populate the in-memory cache from ``geo_cache`` table. Use :meth:`GeoCache.load_cache_from_db` instead. This function delegates to the default module-level instance for backward compatibility only. Args: db: Open :class:`aiosqlite.Connection` to the BanGUI application database (not the fail2ban database). """ await _default_geo_cache.load_cache_from_db(db) async def lookup( ip: str, http_session: aiohttp.ClientSession, db: aiosqlite.Connection | None = None, ) -> GeoInfo | None: """(DEPRECATED) Resolve an IP address to country, ASN, organisation metadata. Use :meth:`GeoCache.lookup` instead. This function delegates to the default module-level instance for backward compatibility only. Args: ip: IPv4 or IPv6 address string. http_session: Shared :class:`aiohttp.ClientSession`. db: Optional BanGUI application database for persistence. Returns: A :class:`GeoInfo` instance, or ``None`` on lookup failure. """ return await _default_geo_cache.lookup(ip, http_session, db=db) def lookup_cached_only( ips: list[str], ) -> tuple[dict[str, GeoInfo], list[str]]: """(DEPRECATED) Return cached geo data without making external API calls. Use :meth:`GeoCache.lookup_cached_only` instead. This function delegates to the default module-level instance for backward compatibility only. Args: ips: IP address strings to look up. Returns: A ``(geo_map, uncached)`` tuple. """ return _default_geo_cache.lookup_cached_only(ips) async def lookup_batch( ips: list[str], http_session: aiohttp.ClientSession, db: aiosqlite.Connection | None = None, ) -> dict[str, GeoInfo]: """(DEPRECATED) Resolve multiple IPs in bulk using ip-api.com batch endpoint. Use :meth:`GeoCache.lookup_batch` instead. This function delegates to the default module-level instance for backward compatibility only. Args: ips: List of IP address strings to resolve. http_session: Shared :class:`aiohttp.ClientSession`. db: Optional BanGUI application database for persistent cache writes. Returns: Dict mapping ``ip → GeoInfo`` for every input IP. """ return await _default_geo_cache.lookup_batch(ips, http_session, db=db) async def flush_dirty(db: aiosqlite.Connection) -> int: """(DEPRECATED) Persist all new in-memory geo entries to the database. Use :meth:`GeoCache.flush_dirty` instead. This function delegates to the default module-level instance for backward compatibility only. Args: db: Open :class:`aiosqlite.Connection` to the BanGUI application database. Returns: The number of rows successfully upserted. """ return await _default_geo_cache.flush_dirty(db) async def re_resolve_all( db: aiosqlite.Connection, http_session: aiohttp.ClientSession, ) -> dict[str, int]: """(DEPRECATED) Retry geo resolution for all unresolved cache entries. Use :meth:`GeoCache.re_resolve_all` instead. This function delegates to the default module-level instance for backward compatibility only. Args: db: BanGUI application database connection. http_session: Shared aiohttp client session. Returns: A dict with ``resolved`` and ``total`` counts. """ return await _default_geo_cache.re_resolve_all(db, http_session)