# -*- 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 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 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