Tomek Wójcik
dcebccf947
Turns out, getting this thing out into the wild isn't as simple as I thought :D Co-authored-by: Tomek Wójcik <labs@tomekwojcik.pl> Co-committed-by: Tomek Wójcik <labs@tomekwojcik.pl>
119 lines
3.1 KiB
Python
119 lines
3.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import logging
|
|
import uuid
|
|
|
|
from django.db import models
|
|
import uuid6
|
|
|
|
from hotpocket_backend.apps.accounts.models import AccessToken
|
|
from hotpocket_backend.apps.core.conf import settings
|
|
from hotpocket_soa.dto.accounts import (
|
|
AccessTokenMetaUpdateIn,
|
|
AccessTokensQuery,
|
|
)
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class AccessTokensService:
|
|
class AccessTokensServiceError(Exception):
|
|
pass
|
|
|
|
class AccessTokenNotFound(AccessTokensServiceError):
|
|
pass
|
|
|
|
def create(self,
|
|
*,
|
|
account_uuid: uuid.UUID,
|
|
origin: str,
|
|
meta: dict,
|
|
) -> AccessToken:
|
|
pk = uuid6.uuid7()
|
|
key = hmac.new(
|
|
settings.SECRET_KEY.encode('ascii'),
|
|
msg=pk.bytes,
|
|
digestmod=hashlib.sha256,
|
|
)
|
|
|
|
return AccessToken.objects.create(
|
|
pk=pk,
|
|
account_uuid=account_uuid,
|
|
key=key.hexdigest(),
|
|
origin=origin,
|
|
meta=meta,
|
|
)
|
|
|
|
def get(self, *, pk: uuid.UUID) -> AccessToken:
|
|
try:
|
|
query_set = AccessToken.active_objects
|
|
|
|
return query_set.get(pk=pk)
|
|
except AccessToken.DoesNotExist as exception:
|
|
raise self.AccessTokenNotFound(
|
|
f'Access Token not found: pk=`{pk}`',
|
|
) from exception
|
|
|
|
def get_by_key(self, *, key: str) -> AccessToken:
|
|
try:
|
|
query_set = AccessToken.active_objects
|
|
|
|
return query_set.get(key=key)
|
|
except AccessToken.DoesNotExist as exception:
|
|
raise self.AccessTokenNotFound(
|
|
f'Access Token not found: key=`{key}`',
|
|
) from exception
|
|
|
|
def search(self,
|
|
*,
|
|
query: AccessTokensQuery,
|
|
offset: int = 0,
|
|
limit: int = 10,
|
|
order_by: str = '-pk',
|
|
) -> models.QuerySet[AccessToken]:
|
|
filters = [
|
|
models.Q(account_uuid=query.account_uuid),
|
|
]
|
|
|
|
if query.before is not None:
|
|
filters.append(models.Q(pk__lt=query.before))
|
|
|
|
result = AccessToken.active_objects.\
|
|
filter(*filters).\
|
|
order_by(order_by)
|
|
|
|
return result[offset:offset + limit]
|
|
|
|
def delete(self, *, pk: uuid.UUID) -> bool:
|
|
access_token = self.get(pk=pk)
|
|
access_token.soft_delete()
|
|
|
|
return True
|
|
|
|
def update_meta(self,
|
|
*,
|
|
pk: uuid.UUID,
|
|
update: AccessTokenMetaUpdateIn,
|
|
) -> AccessToken:
|
|
access_token = AccessToken.active_objects.get(pk=pk)
|
|
|
|
next_meta = {
|
|
**(access_token.meta or {}),
|
|
}
|
|
|
|
if update.version is not None:
|
|
next_meta['version'] = update.version
|
|
|
|
if update.platform is not None:
|
|
next_meta['platform'] = update.platform
|
|
|
|
access_token.meta = next_meta
|
|
access_token.save()
|
|
|
|
access_token.refresh_from_db()
|
|
|
|
return access_token
|