locations.py 文件源码

def get_location_from_geolite2(pool, ip_addr):

    try:
        ip_addr = ipaddress.ip_address(ip_addr)
        # fix issues with asyncpg not handling ip address netmask defaults
        # correctly
        ip_addr = "{}/{}".format(ip_addr, 32 if ip_addr.version == 4 else 128)
        async with pool.acquire() as con:
            row = await con.fetchrow(
                "SELECT cs.country_iso_code FROM geolite2_ip_addresses ips "
                "JOIN geolite2_countries cs ON ips.geoname_id = cs.geoname_id "
                "WHERE ips.network >> $1", ip_addr)
        return row['country_iso_code'] if row else None
    except ValueError:
        return None
    except asyncpg.exceptions.UndefinedTableError:
        log.warning("Missing GeoLite2 database tables")
        return None