class RestApiRepository(Generic[DomainModel]):
"""Implementation of `ApiRepository` that interacts with a RESTful API
using the `requests` library.
This repository provides methods for common CRUD operations (Create, Read,
Update, Delete). It handles the construction of API URLs, serialization/
deserialization of domain models, and interaction with the `requests`
library. It also allows for flexible configuration of the API endpoints,
session management, and response handling. The repository can be easily
extended or customized for specific API requirements.
Generic
-------
The type of the domain model that this repository will manage. This
allows for type safety and better integration with the rest of the
application.
"""
def __init__(
self,
host: str,
scheme: str | None = None,
base_path: str = "",
endpoint: str = "",
default_model: DomainModel | None = None,
use_factory: bool = True,
serialize: bool = True,
model_factory_method_name: str = "from_dict",
model_serialization_method_name: str = "to_dict",
client: HTTPClient | None = None,
session: HTTPClient | None = None,
request_headers: dict[str, str] | None = None,
request_cookies: dict[str, str] | None = None,
response_data_attribute: str | None = None,
) -> None:
"""Initialize the REST API repository.
Parameters
----------
host
The base URL of the API.
scheme
The URL scheme to use (e.g., "http" or "https"). This is only used
if the host does not already include a scheme, by default "https"
base_path
The base path of the API, by default ""
endpoint
The default endpoint for the API. This value is used when no
specific endpoint is provided in the method calls, by default ""
default_model
The default model to use for serialization/deserialization,
by default None
use_factory
Whether to use the model factory method for creating models from
response data, by default True
serialize
Whether to serialize objects before sending them in requests,
by default True
model_factory_method_name
The name of the class method to use for creating models from
dictionaries, by default "from_dict"
model_serialization_method_name
The name of the method to use for serializing models to
dictionaries, by default "to_dict"
client
An HTTP client to use for context management, by default None.
If None, a new `requests.sessions.Session` will be created and
used. This allows for flexibility in using different HTTP client
implementations that conform to the `HTTPClient` protocol, while
still providing a default option with `requests`.
session
Deprecated: An HTTP client session to use for making requests, by
default None. Use `client` instead. This parameter is still
supported for backward compatibility.
request_headers
Default headers to include in every request, by default None
request_cookies
Default cookies to include in every request, by default None
response_data_attribute
The attribute in the response data to extract the relevant data
from, by default None
"""
self._host = host
self._scheme = scheme or "https"
self._base_path = base_path
self._endpoint = endpoint
self._default_model = default_model
self._use_factory = use_factory
self._serialize = serialize
self._model_factory_method_name = model_factory_method_name
self._model_serialization_method_name = model_serialization_method_name
client_obj = client or session or requests.sessions.Session()
# Expose the underlying client publicly for consistency with other
# repositories
self.client = client_obj
# Preserve the deprecated public session alias for backward
# compatibility
self.session = client_obj
# Preserve the existing private attribute for backward compatibility
self._session = client_obj
self._request_headers = request_headers or {}
self._request_cookies = request_cookies or {}
self._response_data_attribute = response_data_attribute
# Update client with default headers and cookies
self.client.headers.update(request_headers or {})
cookiejar_from_dict(
request_cookies or {},
cookiejar=self.client.cookies,
overwrite=True,
)
def add(
self,
obj: DomainModel,
return_obj: bool = True,
serialize: bool | None = None,
use_factory: bool | None = None,
endpoint: str | None = None,
parent_endpoint: str | None = None,
parent_param: str | int | UUID | None = None,
model: DomainModel | None = None,
additional_request_params: dict[str, Any] | None = None,
**params: Any,
) -> DomainModel | dict[str, Any] | None:
"""Add a new resource.
Parameters
----------
obj
The object to add.
return_obj
Whether to return the added object or not.
serialize
Whether to serialize the object before sending it in the API
request.
use_factory
Whether to use the model factory method for creating models from
response data.
endpoint
The API endpoint to which the object should be added.
parent_endpoint
The parent API endpoint, if the resource is nested under a parent
resource.
parent_param
The parameter to identify the parent resource, if applicable. This
could be an ID or a unique identifier. The parameter will be
appended to the parent endpoint to form the full URL for the API
request.
model
The model to use for serialization/deserialization.
additional_request_params
Additional parameters to include in the function call which handles
the API request. This allows for flexibility in specifying
parameters such as headers, authentication tokens, or other request
options that may be needed for the API call.
**params
Additional query parameters to include in the API request.
Returns
-------
The added object if `return_obj` is `True`, otherwise `None`.
"""
if self._determine_serialization(serialize):
obj = self._serialize_object(obj)
url = self._build_url(
endpoint,
parent_endpoint=parent_endpoint,
parent_param=parent_param,
**params,
)
response_data = self._post(
url=url,
data=obj,
additional_request_params=additional_request_params,
)
if return_obj is False:
return None
if not self._determine_use_factory(use_factory):
return response_data
return self._map_response_object(response_data, model)
def add_all(
self,
objs: list[DomainModel],
return_objs: bool = True,
serialize: bool | None = None,
use_factory: bool | None = None,
endpoint: str | None = None,
parent_endpoint: str | None = None,
parent_param: str | int | UUID | None = None,
model: DomainModel | None = None,
additional_request_params: dict[str, Any] | None = None,
one_by_one: bool = False,
**params: Any,
) -> list[DomainModel] | list[dict[str, Any]] | None:
"""Add multiple new resources.
Parameters
----------
objs
The objects to add.
return_objs
Whether to return the added objects or not.
serialize
Whether to serialize the objects before sending it in the API
request.
use_factory
Whether to use the model factory method for creating models from
response data.
endpoint
The API endpoint to which the objects should be added.
parent_endpoint
The parent API endpoint, if the resource is nested under a parent
resource.
parent_param
The parameter to identify the parent resource, if applicable. This
could be an ID or a unique identifier. The parameter will be
appended to the parent endpoint to form the full URL for the API
request.
model
The model to use for serialization/deserialization.
additional_request_params
Additional parameters to include in the function call which handles
the API request. This allows for flexibility in specifying
parameters such as headers, authentication tokens, or other request
options that may be needed for the API call.
one_by_one
Whether to add the objects one by one (i.e. make a separate API
call for each object).
**params
Additional query parameters to include in the API request.
Returns
-------
A list of added objects if `return_objs` is `True`, otherwise
`None`.
"""
if one_by_one:
results: list[DomainModel] | list[dict[str, Any]] = []
for obj in objs:
result = self.add(
obj=obj,
return_obj=return_objs,
serialize=serialize,
use_factory=use_factory,
endpoint=endpoint,
parent_endpoint=parent_endpoint,
parent_param=parent_param,
model=model,
additional_request_params=additional_request_params,
**params,
)
if result is not None:
results.append(result) # type: ignore
return results if return_objs else None
if self._determine_serialization(serialize):
objs = [self._serialize_object(obj) for obj in objs]
url = self._build_url(
endpoint,
parent_endpoint=parent_endpoint,
parent_param=parent_param,
**params,
)
response_data = self._post(
url=url,
data=objs,
additional_request_params=additional_request_params,
)
if return_objs is False:
return None
if not self._determine_use_factory(use_factory):
return response_data
return self._map_response_array(response_data, model)
def get(
self,
use_factory: bool | None = None,
endpoint: str | None = None,
parent_endpoint: str | None = None,
parent_param: str | int | UUID | None = None,
param: str | int | UUID | None = None,
model: DomainModel | None = None,
additional_request_params: dict[str, Any] | None = None,
**params: Any,
) -> DomainModel | dict[str, Any]:
"""Retrieve a single resource.
Parameters
----------
endpoint
The API endpoint from which to retrieve the resource.
use_factory
Whether to use the model factory method for creating models from
response data.
endpoint
The API endpoint to which the object should be added.
parent_endpoint
The parent API endpoint, if the resource is nested under a parent
resource.
parent_param
The parameter to identify the parent resource, if applicable. This
could be an ID or a unique identifier. The parameter will be
appended to the parent endpoint to form the full URL for the API
request.
param
The parameter to identify the specific resource. This could be an
ID or a unique identifier. The parameter will be appended to the
endpoint to form the full URL for the GET request.
model
The model to use for serialization/deserialization.
additional_request_params
Additional parameters to include in the function call which handles
the API request. This allows for flexibility in specifying
parameters such as headers, authentication tokens, or other request
options that may be needed for the API call.
**params
Additional query parameters to include in the API request.
Returns
-------
The retrieved object.
"""
url = self._build_url(
endpoint,
parent_endpoint=parent_endpoint,
parent_param=parent_param,
param=param,
**params,
)
response_data: dict[str, Any] = self._get(
url=url,
additional_request_params=additional_request_params,
)
if not self._determine_use_factory(use_factory):
return response_data
return self._map_response_object(response_data, model)
def get_all(
self,
use_factory: bool | None = None,
endpoint: str | None = None,
parent_endpoint: str | None = None,
parent_param: str | int | UUID | None = None,
param: str | int | UUID | None = None,
model: DomainModel | None = None,
additional_request_params: dict[str, Any] | None = None,
**params: Any,
) -> list[DomainModel] | list[dict[str, Any]]:
"""Retrieve multiple resources.
Parameters
----------
endpoint
The API endpoint from which to retrieve the resource.
use_factory
Whether to use the model factory method for creating models from
response data.
endpoint
The API endpoint to which the object should be added.
parent_endpoint
The parent API endpoint, if the resource is nested under a parent
resource.
parent_param
The parameter to identify the parent resource, if applicable. This
could be an ID or a unique identifier. The parameter will be
appended to the parent endpoint to form the full URL for the API
request.
param
The parameter to identify the specific resource. This could be an
ID or a unique identifier. The parameter will be appended to the
endpoint to form the full URL for the GET request.
model
The model to use for serialization/deserialization.
additional_request_params
Additional parameters to include in the function call which handles
the API request. This allows for flexibility in specifying
parameters such as headers, authentication tokens, or other request
options that may be needed for the API call.
**params
Additional query parameters to include in the API request.
Returns
-------
The retrieved objects.
"""
url = self._build_url(
endpoint,
parent_endpoint=parent_endpoint,
parent_param=parent_param,
param=param,
**params,
)
response_data: list[dict[str, Any]] = self._get(
url=url,
additional_request_params=additional_request_params,
)
if not self._determine_use_factory(use_factory):
return response_data
return self._map_response_array(response_data, model)
def patch(
self,
patch: JsonPatch,
return_obj: bool = True,
use_factory: bool | None = None,
endpoint: str | None = None,
parent_endpoint: str | None = None,
parent_param: str | int | UUID | None = None,
param: str | int | UUID | None = None,
model: DomainModel | None = None,
additional_request_params: dict[str, Any] | None = None,
**params: Any,
) -> DomainModel | dict[str, Any] | None:
"""Update a resource.
Parameters
----------
patch
The JSON Patch object containing the changes to be applied to the
resource. This object should conform to the JSON Patch
specification.
return_obj
Whether to return the updated object or not.
use_factory
Whether to use the model factory method for creating models from
response data.
endpoint
The API endpoint to which the object should be added.
parent_endpoint
The parent API endpoint, if the resource is nested under a parent
resource.
parent_param
The parameter to identify the parent resource, if applicable. This
could be an ID or a unique identifier. The parameter will be
appended to the parent endpoint to form the full URL for the API
request.
param
The parameter to identify the specific resource. This could be an
ID or a unique identifier. The parameter will be appended to the
endpoint to form the full URL for the GET request.
model
The model to use for serialization/deserialization.
additional_request_params
Additional parameters to include in the function call which handles
the API request. This allows for flexibility in specifying
parameters such as headers, authentication tokens, or other request
options that may be needed for the API call.
**params
Additional query parameters to include in the API request.
Returns
-------
The updated object if `return_obj` is `True`, otherwise `None`.
"""
url = self._build_url(
endpoint,
parent_endpoint=parent_endpoint,
parent_param=parent_param,
param=param,
**params,
)
response_data: dict[str, Any] = self._patch(
url=url,
data=patch.patch,
additional_request_params=additional_request_params,
)
if return_obj is False:
return None
if not self._determine_use_factory(use_factory):
return response_data
return self._map_response_object(response_data, model)
def remove(
self,
endpoint: str | None = None,
parent_endpoint: str | None = None,
parent_param: str | int | UUID | None = None,
param: str | int | UUID | None = None,
additional_request_params: dict[str, Any] | None = None,
**params: Any,
) -> None:
"""Remove a resource.
Parameters
----------
endpoint
The API endpoint to which the object should be added.
parent_endpoint
The parent API endpoint, if the resource is nested under a parent
resource.
parent_param
The parameter to identify the parent resource, if applicable. This
could be an ID or a unique identifier. The parameter will be
appended to the parent endpoint to form the full URL for the API
request.
param
The parameter to identify the specific resource. This could be an
ID or a unique identifier. The parameter will be appended to the
endpoint to form the full URL for the GET request.
additional_request_params
Additional parameters to include in the function call which handles
the API request. This allows for flexibility in specifying
parameters such as headers, authentication tokens, or other request
options that may be needed for the API call.
**params
Additional query parameters to include in the API request.
"""
url = self._build_url(
endpoint,
parent_endpoint=parent_endpoint,
parent_param=parent_param,
param=param,
**params,
)
self._delete(
url=url,
additional_request_params=additional_request_params,
)
def update(
self,
obj: DomainModel,
return_obj: bool = True,
serialize: bool | None = None,
use_factory: bool | None = None,
endpoint: str | None = None,
parent_endpoint: str | None = None,
parent_param: str | int | UUID | None = None,
param: str | int | UUID | None = None,
model: DomainModel | None = None,
additional_request_params: dict[str, Any] | None = None,
**params: Any,
) -> DomainModel | dict[str, Any] | None:
"""Update a resource.
Parameters
----------
obj
The object to add.
return_obj
Whether to return the added object or not.
serialize
Whether to serialize the object before sending it in the API
request.
use_factory
Whether to use the model factory method for creating models from
response data.
endpoint
The API endpoint to which the object should be added.
parent_endpoint
The parent API endpoint, if the resource is nested under a parent
resource.
parent_param
The parameter to identify the parent resource, if applicable. This
could be an ID or a unique identifier. The parameter will be
appended to the parent endpoint to form the full URL for the API
request.
param
The parameter to identify the specific resource. This could be an
ID or a unique identifier. The parameter will be appended to the
endpoint to form the full URL for the GET request.
model
The model to use for serialization/deserialization.
additional_request_params
Additional parameters to include in the function call which handles
the API request. This allows for flexibility in specifying
parameters such as headers, authentication tokens, or other request
options that may be needed for the API call.
**params
Additional query parameters to include in the API request.
Returns
-------
The updated object if `return_obj` is `True`, otherwise `None`.
"""
if self._determine_serialization(serialize):
obj = self._serialize_object(obj)
url = self._build_url(
endpoint,
parent_endpoint=parent_endpoint,
parent_param=parent_param,
param=param,
**params,
)
response_data: dict[str, Any] = self._put(
url=url,
data=obj,
additional_request_params=additional_request_params,
)
if return_obj is False:
return None
if not self._determine_use_factory(use_factory):
return response_data
return self._map_response_object(response_data, model)
def _get(
self,
url: str,
additional_request_params: dict[str, Any] | None = None,
) -> Any:
"""Call the GET method of the API.
Parameters
----------
url
A fully constructed URL to which the GET request should be sent.
This URL should include any necessary query parameters. The URL is
expected to be properly formatted and ready for use in an API
request.
additional_request_params
Additional parameters to include in the function call which handles
the API request. This allows for flexibility in specifying
parameters such as headers, authentication tokens, or other request
options that may be needed for the API call.
Returns
-------
The data retrieved from the API response.
"""
response = self.client.get(
url=url,
**(additional_request_params or {}),
)
return self._handle_response(response)
def _post(
self,
url: str,
data: Any,
additional_request_params: dict[str, Any] | None = None,
) -> Any:
"""Call the POST method of the API.
Parameters
----------
url
A fully constructed URL to which the POST request should be sent.
This URL should include any necessary query parameters. The URL is
expected to be properly formatted and ready for use in an API
request.
data
The data to be sent in the body of the POST request. This data is
expected to be in a format that can be serialized to JSON, as it
will be sent as JSON in the request body.
additional_request_params
Additional parameters to include in the function call which handles
the API request. This allows for flexibility in specifying
parameters such as headers, authentication tokens, or other request
options that may be needed for the API call.
Returns
-------
The data retrieved from the API response.
"""
response = self.client.post(
url=url,
json=data,
**(additional_request_params or {}),
)
return self._handle_response(response)
def _patch(
self,
url: str,
data: Any,
additional_request_params: dict[str, Any] | None = None,
) -> Any:
"""Call the PATCH method of the API.
Parameters
----------
url
A fully constructed URL to which the PATCH request should be sent.
This URL should include any necessary query parameters. The URL is
expected to be properly formatted and ready for use in an API
request.
data
The data to be sent in the body of the PATCH request. This data is
expected to be in a format that can be serialized to JSON, as it
will be sent as JSON in the request body.
additional_request_params
Additional parameters to include in the function call which handles
the API request. This allows for flexibility in specifying
parameters such as headers, authentication tokens, or other request
options that may be needed for the API call.
Returns
-------
The data retrieved from the API response.
"""
response = self.client.patch(
url=url,
json=data,
**(additional_request_params or {}),
)
return self._handle_response(response)
def _put(
self,
url: str,
data: Any,
additional_request_params: dict[str, Any] | None = None,
) -> Any:
"""Call the PUT method of the API.
Parameters
----------
url
A fully constructed URL to which the PUT request should be sent.
This URL should include any necessary query parameters. The URL is
expected to be properly formatted and ready for use in an API
request.
data
The data to be sent in the body of the PUT request. This data is
expected to be in a format that can be serialized to JSON, as it
will be sent as JSON in the request body.
additional_request_params
Additional parameters to include in the function call which handles
the API request. This allows for flexibility in specifying
parameters such as headers, authentication tokens, or other request
options that may be needed for the API call.
Returns
-------
The data retrieved from the API response.
"""
response = self.client.put(
url=url,
json=data,
**(additional_request_params or {}),
)
return self._handle_response(response)
def _delete(
self, url: str, additional_request_params: dict[str, Any] | None = None
) -> dict[str, Any] | None:
"""Call the DELETE method of the API.
Parameters
----------
url
A fully constructed URL to which the DELETE request should be sent.
This URL should include any necessary query parameters. The URL is
expected to be properly formatted and ready for use in an API
request.
additional_request_params
Additional parameters to include in the function call which handles
the API request. This allows for flexibility in specifying
parameters such as headers, authentication tokens, or other request
options that may be needed for the API call.
"""
response = self.client.delete(
url=url,
**(additional_request_params or {}),
)
return self._handle_response(response)
def _map_response_object(self, response: Any, model: T | None) -> T:
"""Map a single object from the API response to a model instance.
Parameters
----------
response
The API response containing the data to be mapped.
model
The model class to which the data should be mapped. If None, the
default model for the repository will be used.
Returns
-------
An instance of the model populated with the data from the API
response.
"""
model_to_use = self._determine_model(model)
return getattr(model_to_use, self._model_factory_method_name)(response)
def _map_response_array(
self, response: list[Any], model: T | None
) -> list[T]:
"""Map an array of objects from the API response to model instances.
Parameters
----------
response
The API response containing the data to be mapped.
model
The model class to which the data should be mapped. If None, the
default model for the repository will be used.
Returns
-------
A list of model instances populated with the data from the API
response.
"""
model_to_use = self._determine_model(model)
return [
getattr(model_to_use, self._model_factory_method_name)(item)
for item in response
]
def _build_url(
self,
endpoint: str | None = None,
param: str | int | UUID | None = None,
parent_endpoint: str | None = None,
parent_param: str | int | UUID | None = None,
**params: Any,
) -> str:
"""Build an URL to use for HTTP requests.
The method constructs the URL based on the provided endpoint and
parameter, as well as the base host, scheme, and base path configured
for the repository. The endpoint and parameter are optional, allowing
for flexible URL construction. The method ensures that the URL is
properly formatted and can be used for making API requests.
The method detects if the scheme is provided and constructs the URL
accordingly. If a scheme is provided, it will be included in the URL.
If not, the URL will be constructed without a scheme, allowing for
relative URLs or URLs with a different scheme. If the scheme is already
included in the host, it will not be replaced.
An optional parent endpoint and parameter can be included in the URL,
which is useful for nested resources. The parent parameter will be
appended after the parent endpoint, and the main parameter will be
appended after the main endpoint.
Parameters
----------
endpoint
The endpoint to use for the URL, by default None
param
The parameter to append to the URL, by default None
parent_endpoint
An optional parent endpoint to include in the URL, by default None
parent_param
An optional parameter to append after the parent endpoint,
by default None
**params
Additional parameters that can be used for query parameters.
Returns
-------
The constructed URL as a string
"""
if self._scheme and not self._host.__contains__("://"):
url = f"{self._scheme}://{self._host}"
else:
url = self._host
endpoint = endpoint or self._endpoint
if self._base_path:
url = urljoin(url, self._base_path.strip("/") + "/")
if parent_endpoint:
url = urljoin(url, parent_endpoint.lstrip("/"))
if parent_param is not None:
url = urljoin(url + "/", str(parent_param))
if endpoint:
url = urljoin(url + "/", endpoint.lstrip("/"))
if param is not None:
url = urljoin(url + "/", str(param))
if params:
url = url + "?" + urlencode(params, doseq=True)
return url
def _serialize_object(self, obj: Any) -> Any:
"""Serialize an object using the specified serialization method if it
exists.
Parameters
----------
obj
The object to be serialized.
Returns
-------
The serialized object if the serialization method exists, otherwise
the original object.
"""
if hasattr(obj, self._model_serialization_method_name):
return getattr(obj, self._model_serialization_method_name)()
return obj
def _get_data_from_response(
self, response: Response
) -> dict[str, Any] | None:
"""Extract data from the API response. If the response_data_attribute
is configured, it will return the value of that attribute. Otherwise,
it will return the entire response data.
Parameters
----------
response
The API response object.
Returns
-------
The extracted data from the API response.
"""
data: dict[str, Any] = response.json()
if self._response_data_attribute:
return data.get(self._response_data_attribute)
return data
def _determine_serialization(
self,
serialize: bool | None,
) -> bool:
"""Determine to use the serialize variable or self._serialize for
deciding whether to serialize the object before sending it in the API
request.
Parameters
----------
serialize
Whether to serialize the object before sending it in the API
request.
Returns
-------
The value to use for deciding whether to serialize the object before
sending it in the API request.
"""
return serialize if serialize is not None else self._serialize
def _determine_use_factory(
self,
use_factory: bool | None,
) -> bool:
"""Determine to use the use_factory variable or self._use_factory for
deciding whether to use the model factory method for creating models.
Parameters
----------
use_factory
Whether to use the model factory method for creating models from
response data.
Returns
-------
The value to use for deciding whether to use the model factory
method for creating models from response data.
"""
return use_factory if use_factory is not None else self._use_factory
def _determine_model(self, model: T | None) -> T:
"""Determine the model to use for mapping response data.
Parameters
----------
model
The model class to which the data should be mapped. If None, the
default model for the repository will be used.
Returns
-------
The model class to use for mapping response data.
Raises
------
ValueError
If no model is provided and no default model is set for the
repository.
AttributeError
If the determined model does not have the required factory method
defined.
"""
if not model and not self._default_model:
raise ValueError(
"No model provided for mapping response and no default model "
"set for the repository."
)
model_to_use = cast(T, model or self._default_model)
if not hasattr(model_to_use, self._model_factory_method_name):
raise AttributeError(
f"The model {model_to_use} does not have the factory method "
f"'{self._model_factory_method_name}' defined. Please ensure "
f"that the model has this method or set the correct "
f"model_factory_method_name for the repository."
)
return model_to_use
def _handle_response(self, response: Response) -> Any | None:
"""Handle the API response and extract the relevant data.
In addition to extracting data from successful responses, this method
also handles various HTTP error responses by raising appropriate
exceptions based on the status code of the response. This ensures that
the caller can handle different error scenarios in a structured way.
Parameters
----------
response
The API response object to be processed.
Returns
-------
The processed data extracted from the API response.
Raises
------
exceptions.BadRequestException
If the API response indicates a bad request (HTTP status code 400).
exceptions.UnauthorizedException
If the API response indicates an unauthorized request (HTTP status
code 401).
exceptions.ForbiddenException
If the API response indicates a forbidden request (HTTP status code
403).
exceptions.NotFoundException
If the API response indicates that the requested resource was not
found (HTTP status code 404).
exceptions.MethodNotAllowedException
If the API response indicates that the HTTP method is not allowed
(HTTP status code 405).
exceptions.NotAcceptableException
If the API response indicates that the requested resource is not
acceptable (HTTP status code 406).
exceptions.ConflictException
If the API response indicates a conflict with the current state of
the resource (HTTP status code 409).
exceptions.PayloadTooLargeException
If the API response indicates that the request payload is too large
(HTTP status code 413).
exceptions.UnprocessableContentException
If the API response indicates that the server cannot process the
contained instructions (HTTP status code 422).
exceptions.InternalServerErrorException
If the API response indicates an internal server error (HTTP status
code 500).
exceptions.NotImplementedException
If the API response indicates that the server does not support the
functionality required to fulfill the request (HTTP status code
501).
exceptions.BadGatewayException
If the API response indicates a bad gateway error (HTTP status code
502).
exceptions.ServiceUnavailableException
If the API response indicates that the service is unavailable (HTTP
status code 503).
exceptions.GatewayTimeoutException
If the API response indicates a gateway timeout error (HTTP status
code 504).
exceptions.ClientErrorException
If the API response indicates a client error that is not
specifically handled by the above exceptions.
exceptions.ServerErrorException
If the API response indicates a server error that is not
specifically handled by the above exceptions.
"""
match response.status_code:
case 200 | 201 | 202:
return self._get_data_from_response(response)
case 204:
return None
case 400:
raise exceptions.BadRequestException(
"Bad request: The server could not understand the request due "
"to invalid syntax."
)
case 401:
raise exceptions.UnauthorizedException(
"Unauthorized: The client must authenticate itself to get the "
"requested response."
)
case 403:
raise exceptions.ForbiddenException(
"Forbidden: The client does not have access rights to the "
"content."
)
case 404:
raise exceptions.NotFoundException(
"Not Found: The server can not find the requested resource."
)
case 405:
raise exceptions.MethodNotAllowedException(
"Method Not Allowed: The request method is known by the server "
"but is not supported by the target resource."
)
case 406:
raise exceptions.NotAcceptableException(
"Not Acceptable: The server cannot produce a response matching "
"the list of acceptable values defined in the request's "
"proactive content negotiation headers."
)
case 409:
raise exceptions.ConflictException(
"Conflict: The request could not be completed due to a conflict "
"with the current state of the target resource."
)
case 413:
raise exceptions.PayloadTooLargeException(
"Payload Too Large: The request entity is larger than limits "
"defined by the server."
)
case 422:
raise exceptions.UnprocessableContentException(
"Unprocessable Content: The server understands the content type "
"of the request entity, and the syntax of the request entity is "
"correct, but it was unable to process the contained instructions."
)
case 500:
raise exceptions.InternalServerErrorException(
"Internal Server Error: The server has encountered a situation "
"it doesn't know how to handle."
)
case 501:
raise exceptions.NotImplementedException(
"Not Implemented: The server does not support the functionality "
"required to fulfill the request."
)
case 502:
raise exceptions.BadGatewayException(
"Bad Gateway: The server was acting as a gateway or proxy and "
"received an invalid response from the upstream server."
)
case 503:
raise exceptions.ServiceUnavailableException(
"Service Unavailable: The server is not ready to handle the "
"request. Common causes are a server that is down for "
"maintenance or that is overloaded."
)
case 504:
raise exceptions.GatewayTimeoutException(
"Gateway Timeout: The server was acting as a gateway or proxy and "
"did not receive a timely response from the upstream server."
)
case _:
status_code = response.status_code
if 300 <= status_code < 400:
# Unexpected redirect or other 3xx status not explicitly handled above.
raise exceptions.ClientErrorException(
f"Unexpected redirect or 3xx HTTP status code: {status_code}"
)
if 400 <= status_code < 500:
# Generic client error for 4xx statuses not explicitly handled above.
raise exceptions.ClientErrorException(
f"An HTTP client error occurred (status code {status_code})."
)
if 500 <= status_code < 600:
# Generic server error for 5xx statuses not explicitly handled above.
raise exceptions.ServerErrorException(
f"An HTTP server error occurred (status code {status_code})."
)
# Any other unexpected status code (e.g., 1xx or outside normal ranges).
raise exceptions.ClientErrorException(
f"Unexpected HTTP status code: {status_code}"
)