1  from datetime import timedelta
 2  from enum import Enum
 3  
 4  from httpx import AsyncClient, Response
 5  
 6  from api.logger import get_logger
 7  from api.settings import settings
 8  from api.utils.jwt import encode_jwt
 9  
10  
11  logger = get_logger(__name__)
12  
13  
14  class InternalServiceError(Exception):
15      pass
16  
17  
18  class InternalService(Enum):
19      # SERVICE_XYZ = settings.service_xyz_url
20  
21      def _get_token(self) -> str:
22          return encode_jwt({"aud": self.name.lower()}, timedelta(seconds=settings.internal_jwt_ttl))
23  
24      @classmethod
25      async def _handle_error(cls, response: Response) -> None:
26          if response.status_code in [401, 403] or response.status_code in range(500, 600):
27              await response.aread()
28              raise InternalServiceError(response, response.text)
29  
30      @property
31      def client(self) -> AsyncClient:
32          return AsyncClient(
33              base_url=self.value.rstrip("/") + "/_internal",
34              headers={"Authorization": self._get_token()},
35              event_hooks={"response": [self._handle_error]},
36          )