This repository has been archived by the owner on Aug 2, 2022. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 111
/
base.py
144 lines (121 loc) · 3.93 KB
/
base.py
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
"""Base Service class"""
import base64
import json
import os
from cachecontrol import CacheControl
import requests
from .. import __version__
from mapbox import errors
def Session(access_token=None, env=None):
"""Create an HTTP session.
Parameters
----------
access_token : str
Mapbox access token string (optional).
env : dict, optional
A dict that subsitutes for os.environ.
Returns
-------
requests.Session
"""
if env is None:
env = os.environ.copy()
access_token = (
access_token or
env.get('MapboxAccessToken') or
env.get('MAPBOX_ACCESS_TOKEN'))
session = requests.Session()
session.params.update(access_token=access_token)
session.headers.update({
'User-Agent': 'mapbox-sdk-py/{0} {1}'.format(
__version__, requests.utils.default_user_agent())})
return session
class Service(object):
"""Service base class
Attributes
----------
default_host : str
Default service hostname: api.mapbox.com.
api_name : str
Mapbox API name.
api_version : str
API version string such as "v1" or "v5".
baseuri
username
Methods
-------
handle_http_errors(response, custom_messages=None, raise_for_status=False)
Converts service errors to Python exceptions.
"""
default_host = 'api.mapbox.com'
api_name = 'hors service'
api_version = 'v0'
def __init__(self, access_token=None, host=None, cache=None):
"""Constructs a Service object
This method should be overridden by subclasses.
Parameters
----------
access_token : str
Mapbox access token string.
host : str, optional
Mapbox API host (advanced usage only).
cache : CacheControl cache instance (Dict or FileCache), optional
Optional caching, not generally needed.
Returns
-------
Service
"""
self.session = Session(access_token)
self.host = host or os.environ.get('MAPBOX_HOST', self.default_host)
if cache:
self.session = CacheControl(self.session, cache=cache)
@property
def baseuri(self):
"""The service's base URI
Returns
-------
str
"""
return 'https://{0}/{1}/{2}'.format(
self.host, self.api_name, self.api_version)
@property
def username(self):
"""The username in the service's access token
Returns
-------
str
"""
token = self.session.params.get('access_token')
if not token:
raise errors.TokenError(
"session does not have a valid access_token param")
data = token.split('.')[1]
# replace url chars and add padding
# (https://gist.github.com/perrygeo/ee7c65bb1541ff6ac770)
data = data.replace('-', '+').replace('_', '/') + "==="
try:
return json.loads(base64.b64decode(data).decode('utf-8'))['u']
except (ValueError, KeyError):
raise errors.TokenError(
"access_token does not contain username")
def handle_http_error(self, response, custom_messages=None,
raise_for_status=False):
"""Converts service errors to Python exceptions
Parameters
----------
response : requests.Response
A service response.
custom_messages : dict, optional
A mapping of custom exception messages to HTTP status codes.
raise_for_status : bool, optional
If True, the requests library provides Python exceptions.
Returns
-------
None
"""
if not custom_messages:
custom_messages = {}
if response.status_code in custom_messages.keys():
raise errors.HTTPError(custom_messages[response.status_code])
if raise_for_status:
response.raise_for_status()