from functools import cached_property from typing import Any import httpx from fansifter_common.httpclient.exceptions import HTTPClientError from pydantic import ValidationError from dmp.adapters.meta.constants import ( ERROR_CODE_INVALID_ACCESS_TOKEN, ERROR_CODE_PERMISSIONS_ERROR, ERROR_SUBCODE_ACCESS_CANNOT_VALIDATE_TOKEN, ERROR_SUBCODE_ACCESS_TOKEN_EXPIRED, ERROR_SUBCODE_ACCESS_TOKEN_INVALIDATED, ERROR_SUBCODE_APP_NOT_INSTALLED, ERROR_SUBCODE_TERMS_OF_SERVICE, ) from dmp.adapters.meta.models import Error, ErrorResponse class FacebookClientError(HTTPClientError): def __init__( self, message: str = "Facebook client error", *, request: httpx.Request | None = None, response: httpx.Response | None = None, ): self.message = message self.request = request self.response = response def __str__(self) -> str: return self.message @property def status_code(self) -> int: if self.response: return self.response.status_code return 500 @cached_property def fb_error(self) -> Error | None: if self.response: try: error_response = ErrorResponse.model_validate_json( self.response.content ) except ValidationError: pass else: return error_response.error return None @cached_property def fb_error_data(self) -> dict[str, Any]: if self.fb_error: return self.fb_error.model_dump( exclude_none=True, exclude_unset=True, ) return {} @property def is_permissions_error(self) -> bool: if self.fb_error: return self.fb_error.code == ERROR_CODE_PERMISSIONS_ERROR return False @property def is_terms_of_service_error(self) -> bool: if self.fb_error: return self.fb_error.error_subcode == ERROR_SUBCODE_TERMS_OF_SERVICE return False @property def is_token_validation_error(self) -> bool: if self.fb_error: return ( self.fb_error.code == ERROR_CODE_INVALID_ACCESS_TOKEN and self.fb_error.error_subcode in [ ERROR_SUBCODE_APP_NOT_INSTALLED, ERROR_SUBCODE_ACCESS_TOKEN_INVALIDATED, ERROR_SUBCODE_ACCESS_TOKEN_EXPIRED, ERROR_SUBCODE_ACCESS_CANNOT_VALIDATE_TOKEN, ] ) return False @property def log_msg(self) -> str: if self.fb_error: return self.fb_error.error_user_msg or self.fb_error.message or self.message return self.message @property def log_extra(self) -> dict[str, Any]: extra: dict[str, Any] = {} if self.request: extra["fb_request_method"] = self.request.method extra["fb_request_path"] = self.request.url.path if self.response: extra["fb_response_status_code"] = self.response.status_code if self.fb_error: extra["fb_response_error"] = self.fb_error_data return extra