-
Notifications
You must be signed in to change notification settings - Fork 7
Expand file tree
/
Copy pathclient.py
More file actions
66 lines (56 loc) · 2.33 KB
/
client.py
File metadata and controls
66 lines (56 loc) · 2.33 KB
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
import aiohttp
from .proxy import ResourceProxy
from simple_rest_client.api import API
from simple_rest_client.resource import AsyncResource
from urllib.parse import urlparse
class Client:
def __init__(self, api_name, resources, session=None, api_url=None):
self.api_name = api_name
self.resources = resources
self.session = session
self._custom_api_url = api_url
self._catalog_api_url = None
self._current_api_url = None
self.api = None
if self.session is None:
raise AttributeError("provided session object is None, probably auth error?")
async def init_api(self, timeout=60):
await self.get_credentials()
self.api = API(
api_root_url=self.api_url,
headers={'X-Auth-Token': self.session.token,
'Accept': 'application/json',
'Content-Type': 'application/json'
},
json_encode_body=True,
timeout=timeout
)
for resource in self.resources:
self.api.add_resource(resource_name=resource, resource_class=AsyncResource)
@property
def api_url(self):
api_url = self._custom_api_url or self._current_api_url or self._catalog_api_url
if api_url and not api_url.endswith('/'):
api_url += '/'
return api_url
async def get_current_version_api_url(self, url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
versions = await response.json()
return [version["links"][0]["href"] for version in versions["versions"] if version["status"] == "CURRENT"][0]
async def get_credentials(self):
await self.session.authenticate()
# api_url is provided, don't bother to determine
if self.api_url:
return
# take url from catalog
url = self.session.get_endpoint_url(self.api_name)
parts = urlparse(url)
if parts.path:
# preasumly full url with api or/and project id
self._catalog_api_url = url
else:
# base url so we need to determine full url with version
self._current_api_url = await self.get_current_version_api_url(url)
def __getattr__(self, name):
return ResourceProxy(self.api, name)