Models#

blacksmith.domain.model.params.PathInfoField(default: Any = PydanticUndefined, *, default_factory: Optional[Callable[[], Any]] = None, alias: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, const: Optional[bool] = None, gt: Optional[float] = None, ge: Optional[float] = None, lt: Optional[float] = None, le: Optional[float] = None, multiple_of: Optional[float] = None, allow_inf_nan: Optional[bool] = None, max_digits: Optional[int] = None, decimal_places: Optional[int] = None, min_items: Optional[int] = None, max_items: Optional[int] = None, unique_items: Optional[bool] = None, min_length: Optional[int] = None, max_length: Optional[int] = None, allow_mutation: bool = True, regex: Optional[str] = None, discriminator: Optional[str] = None, repr: bool = True, **extra: Any) → Any#: Declare field that are serialized to the path info.

blacksmith.domain.model.params.HeaderField(default: Any = PydanticUndefined, *, default_factory: Optional[Callable[[], Any]] = None, alias: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, const: Optional[bool] = None, gt: Optional[float] = None, ge: Optional[float] = None, lt: Optional[float] = None, le: Optional[float] = None, multiple_of: Optional[float] = None, allow_inf_nan: Optional[bool] = None, max_digits: Optional[int] = None, decimal_places: Optional[int] = None, min_items: Optional[int] = None, max_items: Optional[int] = None, unique_items: Optional[bool] = None, min_length: Optional[int] = None, max_length: Optional[int] = None, allow_mutation: bool = True, regex: Optional[str] = None, discriminator: Optional[str] = None, repr: bool = True, **extra: Any) → Any#: Declare field that are serialized in http request header.

blacksmith.domain.model.params.QueryStringField(default: Any = PydanticUndefined, *, default_factory: Optional[Callable[[], Any]] = None, alias: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, const: Optional[bool] = None, gt: Optional[float] = None, ge: Optional[float] = None, lt: Optional[float] = None, le: Optional[float] = None, multiple_of: Optional[float] = None, allow_inf_nan: Optional[bool] = None, max_digits: Optional[int] = None, decimal_places: Optional[int] = None, min_items: Optional[int] = None, max_items: Optional[int] = None, unique_items: Optional[bool] = None, min_length: Optional[int] = None, max_length: Optional[int] = None, allow_mutation: bool = True, regex: Optional[str] = None, discriminator: Optional[str] = None, repr: bool = True, **extra: Any) → Any#: Declare field that are serialized in the http querystring.

blacksmith.domain.model.params.PostBodyField(default: Any = PydanticUndefined, *, default_factory: Optional[Callable[[], Any]] = None, alias: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, const: Optional[bool] = None, gt: Optional[float] = None, ge: Optional[float] = None, lt: Optional[float] = None, le: Optional[float] = None, multiple_of: Optional[float] = None, allow_inf_nan: Optional[bool] = None, max_digits: Optional[int] = None, decimal_places: Optional[int] = None, min_items: Optional[int] = None, max_items: Optional[int] = None, unique_items: Optional[bool] = None, min_length: Optional[int] = None, max_length: Optional[int] = None, allow_mutation: bool = True, regex: Optional[str] = None, discriminator: Optional[str] = None, repr: bool = True, **extra: Any) → Any#: Declare field that are serialized in the json document.

blacksmith.domain.model.params.get_location(field: Any) → typing_extensions.Literal[path, headers, querystring, body]#

class blacksmith.domain.model.params.JSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)#

default(o: Any) → Any#

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

blacksmith.domain.model.params.serialize_part(req: blacksmith.domain.model.params.Request, part: Dict[str, Any]) → Dict[str, Union[str, int, float, bool]]#

class blacksmith.domain.model.params.Request#

Request Params Model.

Fields must use subclass PathInfoField(), HeaderField(), QueryStringField() or PostBodyField() to declare each fields.

to_http_request(method: typing_extensions.Literal[HEAD, GET, POST, PUT, PATCH, DELETE, OPTIONS], url_pattern: str) → blacksmith.domain.model.http.HTTPRequest#: Convert the request params to an http request in order to serialize the http request for the client.

class blacksmith.domain.model.params.Response#: Response Model.

class blacksmith.domain.model.params.Metadata(count: int, total_count: Optional[int], links: Dict[Optional[str], Dict[str, str]])#

Metadata of a collection response.

count: int#

total_count: Optional[int]#

links: Dict[Optional[str], Dict[str, str]]#

class blacksmith.domain.model.params.AbstractCollectionParser(resp: blacksmith.domain.model.http.HTTPResponse)#

Signature of the collection parser.

resp: blacksmith.domain.model.http.HTTPResponse#

abstract property meta: blacksmith.domain.model.params.Metadata#

Return the metatadata from the response.

Usually, metadata are in a header, but if the API wrap the list,

{
    "total_items": 0,
    "items": []
}

Then, the Metadata.total_count can be extracted from the json, instead of the header.

abstract property json: List[Any]#

Return the list part of the response the response.

For instance, if an API wrap the list in a structure like

{
    "items": [
        {"objkey": "objval"}
    ]
}

then, the resp.json["items"] has to be returned.

class blacksmith.domain.model.params.CollectionParser(resp: blacksmith.domain.model.http.HTTPResponse)#

Handle the rest collection metadata parser.

Deserialize how a collection is wrapped.

total_count_header: str = 'Total-Count'#

property meta: blacksmith.domain.model.params.Metadata#

Return the metatadata from the response.

Usually, metadata are in a header, but if the API wrap the list,

{
    "total_items": 0,
    "items": []
}

Then, the Metadata.total_count can be extracted from the json, instead of the header.

property json: List[Optional[Any]]#

Return the list part of the response the response.

For instance, if an API wrap the list in a structure like

{
    "items": [
        {"objkey": "objval"}
    ]
}

then, the resp.json["items"] has to be returned.

class blacksmith.domain.model.params.ResponseBox(result: Union[result.result.Ok[blacksmith.domain.model.http.HTTPResponse], result.result.Err[blacksmith.domain.exceptions.HTTPError]], response_schema: Optional[Type[blacksmith.domain.model.params.Response]], method: typing_extensions.Literal[HEAD, GET, POST, PUT, PATCH, DELETE, OPTIONS], path: str, name: str, client_name: str, error_parser: blacksmith.domain.error.AbstractErrorParser[blacksmith.domain.error.TError_co])#

Wrap a HTTP response and deserialize it.

user: ResponseBox[User, HTTPError] = (
    await api.user.get({"username": username})
)
if user.is_ok():
    print(user.unwrap().username)
else:
    print(f"API Call failed: {user.unwrap_err()}")

property json: Optional[Dict[str, Any]]#

Return the raw json response.

It return the raw response body without noticing if its a normal or an error response.

property response: blacksmith.domain.model.params.TResponse#

Parse the response using the schema.

Deprecated since version 2.0: Use ResponseBox.unwrap()

Raises

blacksmith.HTTPError – if the response contains an error.
NoResponseSchemaException – if the response_schema has not been set in the contract.

as_result() → Union[result.result.Ok[blacksmith.domain.model.params.TResponse], result.result.Err[blacksmith.domain.error.TError_co]]#

Return the result as a result.Result.

The blacksmith.ResponseBox mimic the result.Result of the result library, but, you may want to cast the response box as a result.

as_optional() → Union[result.result.Ok[Optional[blacksmith.domain.model.params.TResponse]], result.result.Err[blacksmith.domain.error.TError_co]]#

Expose the instance as an optional result.

In case no response schema has been provided while registering the resource, then a Ok(None) is return to not raise any blacksmith.NoResponseSchemaException

is_ok() → bool#: Return True if the response was an http success.

is_err() → bool#: Return True if the response was an http error.

unwrap() → blacksmith.domain.model.params.TResponse#

Return the parsed response.

Raises: NoResponseSchemaException – if there are no response schema set.

unwrap_err() → blacksmith.domain.error.TError_co#: Return the response error.

unwrap_or(default: blacksmith.domain.model.params.TResponse) → blacksmith.domain.model.params.TResponse#

Return the response or the default value in case of error.

Raises: NoResponseSchemaException – if there are no response schema set.

unwrap_or_else(op: Callable[[blacksmith.domain.error.TError_co], blacksmith.domain.model.params.TResponse]) → blacksmith.domain.model.params.TResponse#

Return the response or the callable return in case of error.

Raises: NoResponseSchemaException – if there are no response schema set.

expect(message: str) → blacksmith.domain.model.params.TResponse#

Return the response or raise an UnwrapError exception with the given message.

Raises: NoResponseSchemaException – if there are no response schema set.

expect_err(message: str) → blacksmith.domain.error.TError_co#: Return the error or raise an UnwrapError exception with the given message.

map(op: Callable[[blacksmith.domain.model.params.TResponse], result.result.U]) → Union[result.result.Ok[result.result.U], result.result.Err[blacksmith.domain.error.TError_co]]#

Apply op on response in case of success, and return the new result.

Raises: NoResponseSchemaException – if there are no response schema set.

map_or(default: result.result.U, op: Callable[[blacksmith.domain.model.params.TResponse], result.result.U]) → result.result.U#

Apply and return op on response in case of success, default in case of error.

Raises: NoResponseSchemaException – if there are no response schema set.

map_or_else(default_op: Callable[[], result.result.U], op: Callable[[blacksmith.domain.model.params.TResponse], result.result.U]) → result.result.U#

Return the result of default_op in case of error otherwise the result of op.

Raises: NoResponseSchemaException – if there are no response schema set.

map_err(op: Callable[[blacksmith.domain.exceptions.HTTPError], result.result.F]) → Union[result.result.Ok[blacksmith.domain.model.params.TResponse], result.result.Err[result.result.F]]#

Apply op on error in case of error, and return the new result.

Raises: NoResponseSchemaException – if there are no response schema set.

and_then(op: Callable[[blacksmith.domain.model.params.TResponse], Union[result.result.Ok[result.result.U], result.result.Err[blacksmith.domain.exceptions.HTTPError]]]) → Union[result.result.Ok[result.result.U], result.result.Err[blacksmith.domain.exceptions.HTTPError]]#

Apply the op function on the response and return it if success

Raises: NoResponseSchemaException – if there are no response schema set.

or_else(op: Callable[[blacksmith.domain.exceptions.HTTPError], Union[result.result.Ok[blacksmith.domain.model.params.TResponse], result.result.Err[result.result.F]]]) → Union[result.result.Ok[blacksmith.domain.model.params.TResponse], result.result.Err[result.result.F]]#

Apply the op function on the error and return it if error

Raises: NoResponseSchemaException – if there are no response schema set.

class blacksmith.domain.model.params.CollectionIterator(response: blacksmith.domain.model.http.HTTPResponse, response_schema: Optional[Type[blacksmith.domain.model.params.Response]], collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser])#

Deserialize the models in a json response list, item by item.

response: blacksmith.domain.model.params.AbstractCollectionParser#

property meta: blacksmith.domain.model.params.Metadata#

Get the response metadata such as counts in http header, links…

Those metadata are generated by the collection_parser.

class blacksmith.domain.model.http.HTTPTimeout(read: float = 30.0, connect: float = 15.0)#

Request timeout.

read: float#

connect: float#

class blacksmith.domain.model.http.HTTPRequest(method: typing_extensions.Literal[HEAD, GET, POST, PUT, PATCH, DELETE, OPTIONS], url_pattern: str, path: Dict[str, Union[str, int, float, bool]] = <factory>, querystring: Dict[str, Union[str, int, float, bool, List[Union[str, int, float, bool]]]] = <factory>, headers: Dict[str, str] = <factory>, body: str = '')#

Internal representation of an http request.

Note that the HTTP method is not present, because the method is the funcion called.

The HTTP Request is filled out using the blacksmith.domain.model.params.Request schema.

method: typing_extensions.Literal[HEAD, GET, POST, PUT, PATCH, DELETE, OPTIONS]#

url_pattern: str#

path: Dict[str, Union[str, int, float, bool]]#

querystring: Dict[str, Union[str, int, float, bool, List[Union[str, int, float, bool]]]]#

headers: Dict[str, str]#

body: str = ''#

property url: str#

blacksmith.domain.model.http.parse_header_links(value: str) → List[Dict[str, str]]#

Returns a list of parsed link headers, for more info see: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Link

The generic syntax of those is:

Link: < uri-reference >; param1=value1; param2="value2"

So for instance:

Link; ‘<http:/…/front.jpeg>; type=”image/jpeg”,<http://…/back.jpeg>;’ would return

[
    {"url": "http:/.../front.jpeg", "type": "image/jpeg"},
    {"url": "http://.../back.jpeg"},
]

Note

Stolen code from httpx _utils.py (private method)

Parameters: value – HTTP Link entity-header field
Returns: list of parsed link headers

class blacksmith.domain.model.http.HTTPResponse(status_code: int, headers: Mapping[str, str], json: Optional[Any])#

Internal representation of an http response.

status_code: int#: HTTP Status code.

headers: Mapping[str, str]#: Header of the response.

json: Optional[Any]#: Json Body of the response.

property links: Dict[Optional[str], Dict[str, str]]#