gfwapiclient.http.endpoints.base

Global Fishing Watch (GFW) API Python Client - Base HTTP EndPoint.

Module Contents

Classes

BaseEndPoint

Base class for an API resource endpoint implementing request handling.

Data

__all__

log

API

gfwapiclient.http.endpoints.base.__all__ = ['BaseEndPoint']
gfwapiclient.http.endpoints.base.log: logging.Logger = 'getLogger(...)'
class gfwapiclient.http.endpoints.base.BaseEndPoint(*, method: http.HTTPMethod, path: str, request_params: Optional[gfwapiclient.http.models.request._RequestParamsT], request_body: Optional[gfwapiclient.http.models.request._RequestBodyT], result_item_class: Type[gfwapiclient.http.models.response._ResultItemT], result_class: Type[gfwapiclient.http.models.response._ResultT], http_client: gfwapiclient.http.client.HTTPClient)

Bases: gfwapiclient.http.endpoints.abc.AbstractBaseEndPoint[gfwapiclient.http.models.request._RequestParamsT, gfwapiclient.http.models.request._RequestBodyT, gfwapiclient.http.models.response._ResultItemT, gfwapiclient.http.models.response._ResultT]

Base class for an API resource endpoint implementing request handling.

This class handles:

  • Building the endpoint request.

  • Sending the request to the API endpoint.

  • Parsing, transforming, and casting successful response data to a result.

  • Parsing, transforming, and casting error responses to errors.

Initialization

Initialize a new BaseEndPoint.

Args:
method (http.HTTPMethod):

The HTTP method used by the endpoint (e.g., GET, POST, PUT, DELETE).

path (str):

The relative path of the API endpoint.

request_params (Optional[_RequestParamsT]):

Query parameters for the request.

request_body (Optional[_RequestBodyT]):

The request body.

result_item_class (Type[_ResultItemT]):

Pydantic model for the expected response item.

result_class (Type[_ResultT]):

Pydantic model for the expected response result.

http_client (HTTPClient):

The HTTP client to send requests.

async request(**kwargs: Any) → gfwapiclient.http.models.response._ResultT

Send an HTTP request for this endpoint.

Args:
**kwargs (Any):

Additional keyword arguments to pass to the httpx.Client.send() method.

Returns:
_ResultT:

The result of the API request as a _ResultT instance.

async _request(**kwargs: Any) → gfwapiclient.http.models.response._ResultT

Perform request-response flow for this endpoint.

Args:
**kwargs (Any):

Additional keyword arguments to pass to the httpx.Client.send() method.

Returns:
_ResultT:

The result of the API request as a _ResultT instance.

Raises:
APITimeoutError:

If the request times out.

APIConnectionError:

If a connection error occurs.

APIStatusError:

If the API returns an HTTP status error.

_process_response_data(*, response: httpx.Response) → gfwapiclient.http.models.response._ResultT

Parse, transform and cast response data.

Args:
response (httpx.Response):

The httpx.Response object to process.

Returns:
_ResultT:

The processed response data as a _ResultT instance.

Raises:
ResultValidationError:

If the response’s Content-Type is invalid.

ResultItemValidationError:

If the response data cannot be casted to the _ResultItemT model.

_parse_response_data(*, response: httpx.Response) → Union[List[Dict[str, Any]], Dict[str, Any]]

Parse response and return data.

Args:
response (httpx.Response):

The httpx.Response object to parse.

Returns:
Union[List[Dict[str, Any]], Dict[str, Any]]:

The parsed response data as a dictionary or a list of dictionaries.

Raises:
ResultValidationError:

If the response’s Content-Type is invalid.

_transform_response_data(*, body: Union[List[Dict[str, Any]], Dict[str, Any]]) → Union[List[Dict[str, Any]], Dict[str, Any]]

Transform and reshape response body and return data.

Args:
body (Union[List[Dict[str, Any]], Dict[str, Any]]):

The parsed response body to transform.

Returns:
Union[List[Dict[str, Any]], Dict[str, Any]]:

The transformed response body as a dictionary or a list of dictionaries.

_cast_response_data(*, body: Union[List[Dict[str, Any]], Dict[str, Any]], response: httpx.Response) → Union[List[gfwapiclient.http.models.response._ResultItemT], gfwapiclient.http.models.response._ResultItemT]

Cast response body and return result item.

Args:
body (Union[List[Dict[str, Any]], Dict[str, Any]]):

The transformed response body to cast.

response (httpx.Response):

The httpx.Response object.

Returns:
Union[List[_ResultItemT], _ResultItemT]:

The casted response data as a _ResultItemT or a list of _ResultItemT.

Raises:
ResultItemValidationError:

If the response data cannot be casted to the _ResultItemT model.

_build_api_result(*, data: Union[List[gfwapiclient.http.models.response._ResultItemT], gfwapiclient.http.models.response._ResultItemT]) → gfwapiclient.http.models.response._ResultT

Build and return result for this API endpoint.

Args:
data (Union[List[_ResultItemT], _ResultItemT]):

The casted response data.

Returns:
_ResultT:

The result of the API request as a _ResultT instance.

_process_api_status_error(*, response: httpx.Response) → gfwapiclient.exceptions.http.APIStatusError

Processes raised HTTP status error.

This function:

  • Parses response data from raised HTTP status error.

  • Transforms it to text or JSON.

  • Casts the raised HTTP status error to a specific APIStatusError subclass.

Args:
response (httpx.Response):

The httpx.Response object representing the error response.

Returns:
APIStatusError:

An APIStatusError instance representing the error.

_cast_api_status_error(*, error_message: str, body: Any, response: httpx.Response) → gfwapiclient.exceptions.http.APIStatusError

Converts raised HTTP status error to specific APIStatusError.

Args:
error_message (str):

The error message.

body (Any):

The error body.

response (httpx.Response):

The httpx.Response object representing the error response.

Returns:
APIStatusError:

An APIStatusError instance representing the error.