53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
|
# standard imports
|
||
|
import hashlib
|
||
|
import urllib.parse
|
||
|
import os
|
||
|
import logging
|
||
|
|
||
|
# external imports
|
||
|
from usumbufu.client.base import (
|
||
|
ClientSession,
|
||
|
BaseTokenStore,
|
||
|
)
|
||
|
from usumbufu.client.bearer import BearerClientSession
|
||
|
|
||
|
logg = logging.getLogger(__name__)
|
||
|
|
||
|
|
||
|
class HTTPSession:
|
||
|
|
||
|
token_dir = '/run/user/{}/clicada/usumbufu/.token'.format(os.getuid())
|
||
|
|
||
|
def __init__(self, url, auth=None):
|
||
|
logg.debug('auth auth {}'.format(auth))
|
||
|
self.base_url = url
|
||
|
url_parts = urllib.parse.urlparse(self.base_url)
|
||
|
self.origin = urllib.parse.urljoin(url_parts[0], url_parts[1])
|
||
|
|
||
|
h = hashlib.sha256()
|
||
|
h.update(self.base_url.encode('utf-8'))
|
||
|
z = h.digest()
|
||
|
|
||
|
token_store_dir = os.path.join(self.token_dir, z.hex())
|
||
|
self.token_store = BaseTokenStore(path=token_store_dir)
|
||
|
|
||
|
self.session = ClientSession(self.origin, token_store=self.token_store)
|
||
|
bearer_handler = BearerClientSession(self.origin, token_store=self.token_store)
|
||
|
self.session.add_subhandler(bearer_handler)
|
||
|
|
||
|
if auth != None:
|
||
|
self.session.add_subhandler(auth)
|
||
|
|
||
|
self.opener = urllib.request.build_opener(self.session)
|
||
|
|
||
|
|
||
|
def open(self, endpoint):
|
||
|
url = urllib.parse.urljoin(self.base_url, endpoint)
|
||
|
logg.debug('open {} with opener {}'.format(url, self))
|
||
|
r = self.opener.open(url)
|
||
|
return r.read().decode('utf-8')
|
||
|
|
||
|
|
||
|
def __str__(self):
|
||
|
return str(self.session)
|