API Documentation

Service

A python class to define a service. Contains a base url for the main HTTP service. Resources can be added to a service.

Parameters:

Name	Type	Description	Default
name	str	Name of the service	required
url	str	Base url of the service	required
resources	Optional[list[Resource]]	A list of resources provided by the service	[]
client	Optional[AsyncClient]	An httpx.AsyncClient instance	None
retry	Optional[int]	Optional argument to specify the number of retries across all resources	None
kwargs	Unpack[HttpxClientInputs]	Additional httpx.AsyncClient parameters. see more	{}

add_resource(resource, client=None, retry=None, **kwargs)

Add a new resource to the service

Parameters:

Name	Type	Description	Default
resource	Resource	The new resource instance	required

request(path, method, **kwargs) async

Helper function to make a request directly from the service level

Note

If you provide path as an empty string or /, this would try to find a root resource definition, if any

Parameters:

Name	Type	Description	Default
path	str	Requested path needs to have the following syntax: /{resource_route}/{handler_route}	required
method	Methods	Requested HTTP Method	required

Resource

A python class used to define a RESTful resource.

Usage

>>> from arrest import Resource

>>> user_resource = Resource(name="user", route="/users", handlers=[("GET", "/")])

Handlers can be provided as a list of tuple, dictionary or instances of ResourceHandler

If provided as a dict or ResourceHandler, the keys / fields have to be set according to the ResourceHandler definiion.

If provided as a tuple, at minimum 2 entries (method, route) or a maximum of 5 entries (method, route, request, response, callback) can be defined.

Parameters:

Name	Type	Description	Default
name	Optional[str]	Unique name of the resource	None
route	Optional[str]	Unique route to the resource	required
response_model	Optional[T]	Pydantic datamodel to wrap the json response	None
handlers	Union[List[ResourceHandler], List[Mapping[str, Any]], List[Tuple[Any, ...]]]	List of handlers	None
client	Optional[AsyncClient]	An httpx.AsyncClient instance	None
retry	Optional[int]	Optional argument to specify the number of retries	None
kwargs	Unpack[HttpxClientInputs]	Additional httpx.AsyncClient parameters, see more	{}

request(method, path, request=None, headers=None, query=None, **kwargs) async

Makes an HTTP request against a handler route

Usage

>>> user_resource.user.request(method="GET")

Parameters:

Name	Type	Description	Default
method	Methods	The HTTP method for the request	required
path	str	Path to a handler specified in the resource	required
request	Union[BaseModel, Mapping[str, Any], None]	A pydantic object containing the necessary fields to make an http request to the handler url Must match the corresponding handler.request pydantic model	None
headers	Optional[Mapping[str, str]]	A multi-dict or httpx.Headers containing additional header key-value pairs	None
query	Optional[Mapping[str, str]]	A multi-dict or httpx.QueryParams containing additional query-param key-value pairs	None
**kwargs		Keyword-arguments matching the path params, if any	{}

Returns:

Name	Type	Description
Response	Any | None	A JSON response in form of a list or dict or, Deserialized into the response pydantic model or, Return value of the callback fn

get(path, request=None, headers=None, query=None, **kwargs) async

Makes a HTTP GET request

see request

post(path, request=None, headers=None, query=None, **kwargs) async

Makes a HTTP POST request

see request

put(path, request=None, headers=None, query=None, **kwargs) async

Makes a HTTP PUT request

see request

patch(path, request=None, headers=None, query=None, **kwargs) async

Makes a HTTP PATCH request

see request

delete(path, request=None, headers=None, query=None, **kwargs) async

Makes a HTTP DELETE request

see request

head(path, request=None, headers=None, query=None, **kwargs) async

Makes a HTTP HEAD request

see request

options(path, request=None, headers=None, query=None, **kwargs) async

Makes a HTTP OPTIONS request

see request

handler(path)

Decorator to bind a custom handler to the resource

Parameters:

Name	Type	Description	Default
path	str	path relative to the current resource	required

Returns:

Type	Description
Any	Any

Httpx Client Arguments

Bases: TypedDict

a typed dict to check for all the necessary fields for building an httpx.AsyncClient instance

Parameters:

• auth - (optional) An authentication class to use when sending requests.
• params - (optional) Query parameters to include in request URLs, as a string, dictionary, or sequence of two-tuples.
• headers - (optional) Dictionary of HTTP headers to include when sending requests.
• cookies - (optional) Dictionary of Cookie items to include when sending requests.
• verify - (optional) SSL certificates (a.k.a CA bundle) used to verify the identity of requested hosts. Either True (default CA bundle), a path to an SSL certificate file, an ssl.SSLContext, or False (which will disable verification).
• cert - (optional) An SSL certificate used by the requested host to authenticate the client. Either a path to an SSL certificate file, or two-tuple of (certificate file, key file), or a three-tuple of (certificate file, key file, password).
• http2 - (optional) A boolean indicating if HTTP/2 support should be enabled. Defaults to False.
• proxies - (optional) A dictionary mapping HTTP protocols to proxy URLs (deprecated).
• mounts - (optional) A dictionary mapping HTTP protocols to proxy URLs
• timeout - (optional) The timeout configuration to use when sending requests.
• follow_redirects - (optional) A boolean indicating whether to follow redirects. See more
• limits - (optional) The limits configuration to use.
• max_redirects - (optional) The maximum number of redirect responses that should be followed.
• event_hooks - (optional) - A dictionary to set event hook callbacks for request and response events. See more
• transport - (optional) A transport class to use for sending requests over the network.
• trust_env - (optional) Enables or disables usage of environment variables for configuration.
• default_encoding - (optional) The default encoding to use for decoding response text, if no charset information is included in a response Content-Type header. Set to a callable for automatic character set detection. Default: "utf-8".

Parameters not included

• base_url - Already used internally in Resource, therefore no need to set it from kwargs
• app - Not required currently as Arrest is primarily built to make external http requests

ResourceHandler

Bases: BaseModel

A pydantic class defining a resource handler

Parameters:

Name	Type	Description	Default
method	Methods	HTTP Method for the handler	required
route	str	Unique path to the handler from its parent resource	required
request	T	Python type to validate the request with	required
response	T	Python type to deserialize the HTTP response	required
callback	Callable	A callable (sync or async) to execute with the HTTP response	required

Exceptions

ArrestError

Bases: BaseException

used in error situations

Source code in arrest/exceptions.py

def __init__(self, *args: object) -> None:
    super().__init__(*args)

base class for all Exception. Used in situations that are not one of the following

ArrestHTTPException

Bases: ArrestError

Source code in arrest/exceptions.py

def __init__(self, status_code: int, data: dict | str) -> None:
    self.status_code = status_code
    self.data = data

used for exceptions during HTTP calls

• .status_code - str status code of the exception, 500 for internal server error
• .data - str json response for the exception

NotFoundException

Bases: ArrestError

Source code in arrest/exceptions.py

def __init__(self, message: str):
    self.message = message

base class for all NotFound-type exceptions

HandlerNotFound

Bases: NotFoundException

Source code in arrest/exceptions.py

def __init__(self, message: str):
    super().__init__(message)

raised when no matching handler is found for the requested path

• .message - str

ResourceNotFound

Bases: NotFoundException

Source code in arrest/exceptions.py

def __init__(self, message: str):
    super().__init__(message)

raised when no matching resource is found for the service

• .message - str

ConversionError

Bases: ArrestError

Source code in arrest/exceptions.py

def __init__(self, *args: object) -> None:
    super().__init__(*args)

raised when Arrest cannot convert path-parameter type using any of the existing converters

OpenAPIGenerator

class for generating Arrest services, resources and schema components from OpenAPI specification (>= v3.0) Generates three files:

1. models.py (contains OpenAPI Schema component definitions)
2. resources.py (Arrest Resources based on the path items)
3. services.py (Arrest Services using the resources)

Parameters:

Name	Type	Description	Default
url	str	an HTTP url or full path to the OpenAPI specification (json or yaml)	required
output_path	str	path where the generated files will be saved	required
dir_name	Optional[str]	(optional) specify the folder name containing the files	None
use_pydantic_v2	Optional[bool]	(optional) use pydantic v2 (default: False)	False

Source code in arrest/openapi/parser.py

def __init__(
    self,
    *,
    url: str,
    output_path: str,
    dir_name: Optional[str] = None,
    use_pydantic_v2: Optional[bool] = False,
) -> None:
    """
    class for generating Arrest services, resources and schema
    components from OpenAPI specification (>= v3.0)
    Generates three files:

    1. models.py (contains OpenAPI Schema component definitions)
    2. resources.py (Arrest Resources based on the path items)
    3. services.py (Arrest Services using the resources)


    Parameters:
        url:
            an HTTP url or full path to the OpenAPI specification (json or yaml)
        output_path:
            path where the generated files will be saved
        dir_name:
            (optional) specify the folder name containing the files
        use_pydantic_v2:
            (optional) use pydantic v2 (default: False)

    """
    self.url: str = url
    self.output_path: str = output_path
    self.dir_name: str = dir_name
    self.use_pydantic_v2 = use_pydantic_v2

generate_schema(fmt=None)

Generates the boilerplate files against an OpenAPI Spec

Parameters:

Name	Type	Description	Default
fmt	Optional[Format]	specification format [json, yaml, yml]	None

Raises:

Type	Description
ArrestError	if the output path does not exist

Source code in arrest/openapi/parser.py

def generate_schema(self, fmt: Optional[Format] = None):
    """Generates the boilerplate files against an OpenAPI Spec

    Parameters:
        fmt (Optional[Format], optional): specification format [json, yaml, yml]

    Raises:
        ArrestError: if the output path does not exist
    """
    openapi_bytes = self.download_openapi_spec()
    fmt = fmt if fmt else self.url.split(".")[-1]
    openapi: OpenAPI = self.parse_openapi(fmt=fmt, data=io.BytesIO(openapi_bytes))

    output_path = Path(self.output_path)
    if not output_path.exists():
        raise ArrestError("output path does not exist")

    service_name = self.dir_name or self.get_service_name(openapi)
    output_path = output_path / service_name
    schema_path = output_path / OPENAPI_SCHEMA_FILENAME

    Path.mkdir(output_path, exist_ok=True)

    self.generate_component_schema(input_bytes=openapi_bytes, schema_path=schema_path)
    resources = self.generate_resource_file(
        openapi=openapi, schema_path=schema_path, resource_path=output_path
    )
    self.generate_service_file(openapi=openapi, service_path=output_path, resources=resources)
    InitTemplate(destination_path=output_path).render_and_save()