python/ipinfo/handler_utils.py at cc50fe88a24275fad9b0e63a278d6139a222d989 · abstract-333/python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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
"""
Utilities used in handlers.
"""

import copy
import json
import os
import sys

from .version import __version__ as SDK_VERSION

# Base URL to make requests against.
API_URL = "https://ipinfo.io"

# Base URL for the IPinfo Lite API
LITE_API_URL = "https://api.ipinfo.io/lite"

# Base URL for the IPinfo Core API
CORE_API_URL = "https://api.ipinfo.io/lookup"

# Base URL for the IPinfo Plus API (same as Core)
PLUS_API_URL = "https://api.ipinfo.io/lookup"

# Base URL for the IPinfo Residential Proxy API
RESPROXY_API_URL = "https://ipinfo.io/resproxy"

# Base URL to get country flag image link.
# "PK" -> "https://cdn.ipinfo.io/static/images/countries-flags/PK.svg"
COUNTRY_FLAGS_URL = "https://cdn.ipinfo.io/static/images/countries-flags/"

# The max amount of IPs allowed by the API per batch request.
BATCH_MAX_SIZE = 1000

# The default max size of the cache in terms of number of items.
CACHE_MAXSIZE = 4096

# The default TTL of the cache in seconds
CACHE_TTL = 60 * 60 * 24

# The current version of the cached data.
# Update this if the data being cached has changed in shape for the same key.
CACHE_KEY_VSN = "1"

# The default request timeout for per-IP requests.
REQUEST_TIMEOUT_DEFAULT = 2

# The default request timeout for batch requests.
BATCH_REQ_TIMEOUT_DEFAULT = 5


def get_headers(access_token, custom_headers):
    """Build headers for request to IPinfo API."""
    headers = {
        "user-agent": "IPinfoClient/Python{version}/{sdk_version}".format(
            version=sys.version_info[0], sdk_version=SDK_VERSION
        ),
        "accept": "application/json",
    }

    if custom_headers:
        headers = {**headers, **custom_headers}

    if access_token:
        headers["authorization"] = "Bearer {}".format(access_token)

    return headers


def format_details(
    details,
    countries,
    eu_countries,
    countries_flags,
    countries_currencies,
    continents,
):
    """
    Format details given a countries object.
    """
    country_code = ""
    # Core and Lite API return the country_code in differently named fields
    if "country_code" in details:
        country_code = details.get("country_code")
    elif "country" in details:
        country_code = details.get("country")

    # country_code = details.get("country")
    if country_name := countries.get(country_code):
        details["country_name"] = country_name
    details["isEU"] = country_code in eu_countries
    details["country_flag_url"] = COUNTRY_FLAGS_URL + country_code + ".svg"
    if flag := countries_flags.get(country_code):
        details["country_flag"] = copy.deepcopy(flag)
    if currency := countries_currencies.get(country_code):
        details["country_currency"] = copy.deepcopy(currency)
    if continent := continents.get(country_code):
        details["continent"] = copy.deepcopy(continent)
    if location := details.get("loc"):
        details["latitude"], details["longitude"] = read_coords(location)


def read_coords(location):
    """
    Given a location of the form `<lat>,<lon>`, returns the latitude and
    longitude as a tuple.

    Returns None for each tuple item if the form is invalid.
    """
    lat, lon = None, None
    coords = tuple(location.split(",")) if location else ""
    if len(coords) == 2 and coords[0] and coords[1]:
        lat, lon = coords[0], coords[1]
    return lat, lon


def read_json_file(json_file):
    json_file = os.path.join(os.path.dirname(__file__), json_file)
    with open(json_file, encoding="utf8") as f:
        json_data = f.read()

    return json.loads(json_data)


def return_or_fail(raise_on_fail, e, v):
    """
    Either throws `e` if `raise_on_fail` or else returns `v`.
    """
    if raise_on_fail:
        raise e
    else:
        return v


def cache_key(k):
    """
    Transforms a user-input key into a versioned cache key.
    """
    return f"{k}:{CACHE_KEY_VSN}"


def is_prefixed_lookup(ip_address):
    """
    Check if the address is a prefixed batch lookup (e.g., "resproxy/1.2.3.4",
    "lookup/8.8.8.8", "domains/google.com").

    Prefixed lookups skip bogon checking as they are not plain IP addresses.
    """
    return isinstance(ip_address, str) and "/" in ip_address