class ApiEndpoint(Generic[ResponseBodyT]):
"""
Descriptor for defining API endpoints on a client.
Allows us to share state (like auth or HTTP client) between API endpoints
of the same client.
See `BaseClient` for an example of how to use this.
"""
def __init__(self, cls: type[BaseApi[ResponseBodyT]]):
self._api: BaseApi[ResponseBodyT] | None = None
self._api_cls = cls
if self._api_cls is not None and not (issubclass(self._api_cls, BaseApi)):
raise ClientSetupError(attribute="cls")
def __set_name__(self, owner: type[BaseClient] | None, field_name: str) -> None:
self._field_name = field_name
@overload
def __get__(
self, instance: None, owner: type[BaseClient] | None
) -> type[BaseApi[ResponseBodyT]]: ...
@overload
def __get__(self, instance: BaseClient, owner: type[BaseClient] | None) -> Self: ...
def __get__(self, instance, owner): # type: ignore [no-untyped-def]
if instance is None:
# Client has not been initialized, return the API class.
return self._api_cls
if self._api is None:
# If client has been initialized, also initialize the API class.
self._api = self._api_cls(
base_url=instance.base_url,
http_client=instance.http_client,
auth=instance.auth,
)
return self
def __call__(
self,
request_params: "DictSerializableT | None" = None,
request_body: "DictSerializableT | None" = None,
http_client: BaseHttpClient | None = None,
auth: BaseHttpClientAuth = USE_DEFAULT,
) -> BaseResponse[ResponseBodyT]:
if self._api is None:
raise AttributeError("API endpoint not part of a `BaseClient` instance.") # noqa: TRY003
return self._api.execute(
request_params=request_params,
request_body=request_body,
http_client=http_client,
auth=auth,
)
def __set__(self, instance: BaseClient, value: Any) -> NoReturn:
raise AttributeError( # noqa: TRY003
f"`{self._field_name}` is read-only and cannot be modified.",
name=self._field_name,
)
def __delete__(self, instance: BaseClient) -> NoReturn:
raise AttributeError( # noqa: TRY003
f"`{self._field_name}` is read-only and cannot be deleted.",
name=self._field_name,
)