hotpocket/services/packages/soa/hotpocket_soa/services/access_tokens.py

# -*- coding: utf-8 -*-
from __future__ import annotations

import http
import uuid

from hotpocket_backend.apps.accounts.services import (
    AccessTokensService as BackendAccessTokensService,
)
from hotpocket_soa.dto.accounts import (
    AccessTokenMetaUpdateIn,
    AccessTokenOut,
    AccessTokensQuery,
)
from hotpocket_soa.exceptions.backend import NotFound

from .base import ProxyService, SOAError


class AccessTokensService(ProxyService):
    class AccessTokensServiceError(SOAError):
        pass

    class NotFound(AccessTokensServiceError):
        pass

    class AccessDenied(AccessTokensServiceError):
        pass

    def __init__(self):
        super().__init__()
        self.backend_access_tokens_service = BackendAccessTokensService()

    def get_error_class(self) -> type[SOAError]:
        return self.AccessTokensServiceError

    def create(self,
               *,
               account_uuid: uuid.UUID,
               origin: str,
               meta: dict,
               ) -> AccessTokenOut:
        return AccessTokenOut.model_validate(
            self.call(
                self.backend_access_tokens_service,
                'create',
                account_uuid=account_uuid,
                origin=origin,
                meta=meta,
            ),
            from_attributes=True,
        )

    def get(self,
            *,
            account_uuid: uuid.UUID,
            pk: uuid.UUID,
            ) -> AccessTokenOut:
        try:
            result = AccessTokenOut.model_validate(
                self.call(
                    self.backend_access_tokens_service,
                    'get',
                    pk=pk,
                ),
                from_attributes=True,
            )

            if result.account_uuid != account_uuid:
                raise self.AccessDenied(
                    http.HTTPStatus.FORBIDDEN.value,
                    f'account_uuid=`{account_uuid}` pk=`{pk}`',
                )

            return result
        except NotFound as exception:
            raise self.NotFound.from_backend_error(exception)

    def get_by_key(self,
                   *,
                   account_uuid: uuid.UUID | None,
                   key: str,
                   ) -> AccessTokenOut:
        try:
            result = AccessTokenOut.model_validate(
                self.call(
                    self.backend_access_tokens_service,
                    'get_by_key',
                    key=key,
                ),
                from_attributes=True,
            )

            if result.account_uuid != account_uuid:
                raise self.AccessDenied(
                    http.HTTPStatus.FORBIDDEN.value,
                    f'account_uuid=`{account_uuid}` key=`{key}`',
                )

            return result
        except NotFound as exception:
            raise self.NotFound.from_backend_error(exception)

    def search(self,
               *,
               query: AccessTokensQuery,
               limit: int,
               ) -> list[AccessTokenOut]:
        return [
            AccessTokenOut.model_validate(row, from_attributes=True)
            for row
            in self.call(
                self.backend_access_tokens_service,
                'search',
                query=query,
                limit=limit,
            )
        ]

    def delete(self, *, access_token: AccessTokenOut) -> bool:
        try:
            return self.call(
                self.backend_access_tokens_service,
                'delete',
                pk=access_token.pk,
            )
        except NotFound as exception:
            raise self.NotFound.from_backend_error(exception)

    def update_meta(self,
                    *,
                    access_token: AccessTokenOut,
                    update: AccessTokenMetaUpdateIn,
                    ) -> AccessTokenOut:
        try:
            return AccessTokenOut.model_validate(
                self.call(
                    self.backend_access_tokens_service,
                    'update_meta',
                    pk=access_token.pk,
                    update=update,
                ),
                from_attributes=True,
            )
        except NotFound as exception:
            raise self.NotFound.from_backend_error(exception)