91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
from ..exceptions import ensureKey
|
|
from ..time import Time
|
|
from typing import Optional
|
|
from datetime import datetime, timedelta
|
|
|
|
KEY_REFRESH_TOKEN = 'refresh_token'
|
|
KEY_CLIENT_ID = 'client_id'
|
|
KEY_CLIENT_SECRET = 'client_secret'
|
|
KEY_EXPIRES_IN = 'expires_in'
|
|
KEY_TOKEN_EXPIRY = 'token_expiry'
|
|
KEY_ACCESS_TOKEN = 'access_token'
|
|
|
|
|
|
class Creds():
|
|
def __init__(self, time: Time, id: str, expiration: datetime,
|
|
access_token: str, refresh_token: str,
|
|
secret: Optional[str] = None, original_expiration: datetime = None):
|
|
self._id = id
|
|
self.time: Time = time
|
|
self._secret = secret
|
|
self._access_token = access_token
|
|
self._refresh_token = refresh_token
|
|
self._expiration = expiration
|
|
self._original_expiration = original_expiration
|
|
|
|
@property
|
|
def id(self):
|
|
return self._id
|
|
|
|
@property
|
|
def secret(self):
|
|
return self._secret
|
|
|
|
@property
|
|
def refresh_token(self):
|
|
return self._refresh_token
|
|
|
|
@property
|
|
def access_token(self):
|
|
return self._access_token
|
|
|
|
@property
|
|
def expiration(self):
|
|
if self._expiration is None:
|
|
return self.time.now()
|
|
return self._expiration
|
|
|
|
@property
|
|
def original_expiration(self) -> datetime:
|
|
return self._original_expiration
|
|
|
|
@property
|
|
def is_expired(self):
|
|
return self.time.now() >= self.expiration
|
|
|
|
def serialize(self, include_secret=True):
|
|
ret = {
|
|
"client_id": self.id
|
|
}
|
|
if self.secret is not None and include_secret:
|
|
ret[KEY_CLIENT_SECRET] = self.secret
|
|
if self.refresh_token is not None:
|
|
ret[KEY_REFRESH_TOKEN] = self.refresh_token
|
|
if self.access_token is not None:
|
|
ret[KEY_ACCESS_TOKEN] = self.access_token
|
|
if self.expiration is not None:
|
|
ret[KEY_TOKEN_EXPIRY] = self.time.asRfc3339String(self.expiration)
|
|
return ret
|
|
|
|
@classmethod
|
|
def load(cls, time: Time, data, id=None, secret=None, original_expiration=None):
|
|
if id is None:
|
|
id = ensureKey(KEY_CLIENT_ID, data, "credentials")
|
|
if secret is None and KEY_CLIENT_SECRET in data:
|
|
secret = data[KEY_CLIENT_SECRET]
|
|
refresh = ensureKey(KEY_REFRESH_TOKEN, data, "credentials")
|
|
access = ensureKey(KEY_ACCESS_TOKEN, data, "credentials")
|
|
expires = None
|
|
try:
|
|
if KEY_TOKEN_EXPIRY in data:
|
|
expires = time.parse(data[KEY_TOKEN_EXPIRY])
|
|
if original_expiration is None:
|
|
original_expiration = expires
|
|
elif KEY_EXPIRES_IN in data:
|
|
expires = time.now() + timedelta(seconds=int(data[KEY_EXPIRES_IN]))
|
|
else:
|
|
expires = time.now()
|
|
except BaseException:
|
|
expires = time.now()
|
|
return Creds(time=time, id=id, access_token=access, refresh_token=refresh, secret=secret, expiration=expires, original_expiration=original_expiration)
|