Models

blacksmith.domain.model.params.PathInfoField(default: Any = PydanticUndefined, *, default_factory: typing.Callable[[], Any] | None = PydanticUndefined, alias: str | None = PydanticUndefined, alias_priority: int | None = PydanticUndefined, validation_alias: str | AliasPath | AliasChoices | None = PydanticUndefined, serialization_alias: str | None = PydanticUndefined, title: str | None = PydanticUndefined, description: str | None = PydanticUndefined, examples: list[Any] | None = PydanticUndefined, exclude: bool | None = PydanticUndefined, discriminator: str | types.Discriminator | None = PydanticUndefined, json_schema_extra: JsonDict | typing.Callable[[JsonDict], None] | None = PydanticUndefined, frozen: bool | None = PydanticUndefined, validate_default: bool | None = PydanticUndefined, repr: bool = PydanticUndefined, init_var: bool | None = PydanticUndefined, kw_only: bool | None = PydanticUndefined, pattern: str | None = PydanticUndefined, strict: bool | None = PydanticUndefined, gt: float | None = PydanticUndefined, ge: float | None = PydanticUndefined, lt: float | None = PydanticUndefined, le: float | None = PydanticUndefined, multiple_of: float | None = PydanticUndefined, allow_inf_nan: bool | None = PydanticUndefined, max_digits: int | None = PydanticUndefined, decimal_places: int | None = PydanticUndefined, min_length: int | None = PydanticUndefined, max_length: int | None = PydanticUndefined, union_mode: Literal['smart', 'left_to_right'] = PydanticUndefined, **extra: Unpack[_EmptyKwargs]) → Any

Declare field that are serialized to the path info.

blacksmith.domain.model.params.HeaderField(default: Any = PydanticUndefined, *, default_factory: typing.Callable[[], Any] | None = PydanticUndefined, alias: str | None = PydanticUndefined, alias_priority: int | None = PydanticUndefined, validation_alias: str | AliasPath | AliasChoices | None = PydanticUndefined, serialization_alias: str | None = PydanticUndefined, title: str | None = PydanticUndefined, description: str | None = PydanticUndefined, examples: list[Any] | None = PydanticUndefined, exclude: bool | None = PydanticUndefined, discriminator: str | types.Discriminator | None = PydanticUndefined, json_schema_extra: JsonDict | typing.Callable[[JsonDict], None] | None = PydanticUndefined, frozen: bool | None = PydanticUndefined, validate_default: bool | None = PydanticUndefined, repr: bool = PydanticUndefined, init_var: bool | None = PydanticUndefined, kw_only: bool | None = PydanticUndefined, pattern: str | None = PydanticUndefined, strict: bool | None = PydanticUndefined, gt: float | None = PydanticUndefined, ge: float | None = PydanticUndefined, lt: float | None = PydanticUndefined, le: float | None = PydanticUndefined, multiple_of: float | None = PydanticUndefined, allow_inf_nan: bool | None = PydanticUndefined, max_digits: int | None = PydanticUndefined, decimal_places: int | None = PydanticUndefined, min_length: int | None = PydanticUndefined, max_length: int | None = PydanticUndefined, union_mode: Literal['smart', 'left_to_right'] = PydanticUndefined, **extra: Unpack[_EmptyKwargs]) → Any

Declare field that are serialized in http request header.

blacksmith.domain.model.params.QueryStringField(default: Any = PydanticUndefined, *, default_factory: typing.Callable[[], Any] | None = PydanticUndefined, alias: str | None = PydanticUndefined, alias_priority: int | None = PydanticUndefined, validation_alias: str | AliasPath | AliasChoices | None = PydanticUndefined, serialization_alias: str | None = PydanticUndefined, title: str | None = PydanticUndefined, description: str | None = PydanticUndefined, examples: list[Any] | None = PydanticUndefined, exclude: bool | None = PydanticUndefined, discriminator: str | types.Discriminator | None = PydanticUndefined, json_schema_extra: JsonDict | typing.Callable[[JsonDict], None] | None = PydanticUndefined, frozen: bool | None = PydanticUndefined, validate_default: bool | None = PydanticUndefined, repr: bool = PydanticUndefined, init_var: bool | None = PydanticUndefined, kw_only: bool | None = PydanticUndefined, pattern: str | None = PydanticUndefined, strict: bool | None = PydanticUndefined, gt: float | None = PydanticUndefined, ge: float | None = PydanticUndefined, lt: float | None = PydanticUndefined, le: float | None = PydanticUndefined, multiple_of: float | None = PydanticUndefined, allow_inf_nan: bool | None = PydanticUndefined, max_digits: int | None = PydanticUndefined, decimal_places: int | None = PydanticUndefined, min_length: int | None = PydanticUndefined, max_length: int | None = PydanticUndefined, union_mode: Literal['smart', 'left_to_right'] = PydanticUndefined, **extra: Unpack[_EmptyKwargs]) → Any

Declare field that are serialized in the http querystring.

blacksmith.domain.model.params.PostBodyField(default: Any = PydanticUndefined, *, default_factory: typing.Callable[[], Any] | None = PydanticUndefined, alias: str | None = PydanticUndefined, alias_priority: int | None = PydanticUndefined, validation_alias: str | AliasPath | AliasChoices | None = PydanticUndefined, serialization_alias: str | None = PydanticUndefined, title: str | None = PydanticUndefined, description: str | None = PydanticUndefined, examples: list[Any] | None = PydanticUndefined, exclude: bool | None = PydanticUndefined, discriminator: str | types.Discriminator | None = PydanticUndefined, json_schema_extra: JsonDict | typing.Callable[[JsonDict], None] | None = PydanticUndefined, frozen: bool | None = PydanticUndefined, validate_default: bool | None = PydanticUndefined, repr: bool = PydanticUndefined, init_var: bool | None = PydanticUndefined, kw_only: bool | None = PydanticUndefined, pattern: str | None = PydanticUndefined, strict: bool | None = PydanticUndefined, gt: float | None = PydanticUndefined, ge: float | None = PydanticUndefined, lt: float | None = PydanticUndefined, le: float | None = PydanticUndefined, multiple_of: float | None = PydanticUndefined, allow_inf_nan: bool | None = PydanticUndefined, max_digits: int | None = PydanticUndefined, decimal_places: int | None = PydanticUndefined, min_length: int | None = PydanticUndefined, max_length: int | None = PydanticUndefined, union_mode: Literal['smart', 'left_to_right'] = PydanticUndefined, **extra: Unpack[_EmptyKwargs]) → Any

Declare field that are serialized in the json document.

class blacksmith.domain.model.params.Request

Request Params Model.

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

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class blacksmith.domain.model.params.Response

Response Model.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

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: 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.

unwrap_or_raise(exc: Type[Exception]) → blacksmith.domain.model.params.TResponse

Return the response or raise the exception exc.

Raises

• exc – it the response is an error.

• 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: 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: Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]] = '')

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: 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: Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]] = ''

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.HTTPRawResponse(*args, **kwargs)

Internal representation of an http response. This format is used to deserialize the response body to the HTTPResponse.

status_code: int

headers: Mapping[str, str]

The headers response implmentation should be key insensitive, http standard.

Blacksmith rely on httpx as the default implementation, key are insensitive.

property content: bytes

property text: str

property encoding: str

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

Intermediate representation of an http response.

In this representation, the response body has been parsed to the property json, which is a python structure containing simple python types. This http response representation will be used create pydantic response object.

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]]