Models

blacksmith.domain.model.params.PathInfoField(default: Any = PydanticUndefined, *, default_factory: Callable[[], Any] | Callable[[dict[str, Any]], 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, field_title_generator: Callable[[str, FieldInfo], str] | None = PydanticUndefined, description: str | None = PydanticUndefined, examples: list[Any] | None = PydanticUndefined, exclude: bool | None = PydanticUndefined, discriminator: str | types.Discriminator | None = PydanticUndefined, deprecated: Deprecated | str | bool | None = PydanticUndefined, json_schema_extra: JsonDict | Callable[[JsonDict], None] | None = {'location': 'path'}, frozen: bool | None = PydanticUndefined, validate_default: bool | None = PydanticUndefined, repr: bool = PydanticUndefined, init: bool | None = PydanticUndefined, init_var: bool | None = PydanticUndefined, kw_only: bool | None = PydanticUndefined, pattern: str | Pattern[str] | None = PydanticUndefined, strict: bool | None = PydanticUndefined, coerce_numbers_to_str: bool | None = PydanticUndefined, gt: annotated_types.SupportsGt | None = PydanticUndefined, ge: annotated_types.SupportsGe | None = PydanticUndefined, lt: annotated_types.SupportsLt | None = PydanticUndefined, le: annotated_types.SupportsLe | 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, fail_fast: bool | None = PydanticUndefined, **extra: Unpack[_EmptyKwargs]) → Any

Declare field that are serialized to the path info.

blacksmith.domain.model.params.HeaderField(default: Any = PydanticUndefined, *, default_factory: Callable[[], Any] | Callable[[dict[str, Any]], 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, field_title_generator: Callable[[str, FieldInfo], str] | None = PydanticUndefined, description: str | None = PydanticUndefined, examples: list[Any] | None = PydanticUndefined, exclude: bool | None = PydanticUndefined, discriminator: str | types.Discriminator | None = PydanticUndefined, deprecated: Deprecated | str | bool | None = PydanticUndefined, json_schema_extra: JsonDict | Callable[[JsonDict], None] | None = {'location': 'headers'}, frozen: bool | None = PydanticUndefined, validate_default: bool | None = PydanticUndefined, repr: bool = PydanticUndefined, init: bool | None = PydanticUndefined, init_var: bool | None = PydanticUndefined, kw_only: bool | None = PydanticUndefined, pattern: str | Pattern[str] | None = PydanticUndefined, strict: bool | None = PydanticUndefined, coerce_numbers_to_str: bool | None = PydanticUndefined, gt: annotated_types.SupportsGt | None = PydanticUndefined, ge: annotated_types.SupportsGe | None = PydanticUndefined, lt: annotated_types.SupportsLt | None = PydanticUndefined, le: annotated_types.SupportsLe | 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, fail_fast: bool | None = PydanticUndefined, **extra: Unpack[_EmptyKwargs]) → Any

Declare field that are serialized in http request header.

blacksmith.domain.model.params.QueryStringField(default: Any = PydanticUndefined, *, default_factory: Callable[[], Any] | Callable[[dict[str, Any]], 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, field_title_generator: Callable[[str, FieldInfo], str] | None = PydanticUndefined, description: str | None = PydanticUndefined, examples: list[Any] | None = PydanticUndefined, exclude: bool | None = PydanticUndefined, discriminator: str | types.Discriminator | None = PydanticUndefined, deprecated: Deprecated | str | bool | None = PydanticUndefined, json_schema_extra: JsonDict | Callable[[JsonDict], None] | None = {'location': 'querystring'}, frozen: bool | None = PydanticUndefined, validate_default: bool | None = PydanticUndefined, repr: bool = PydanticUndefined, init: bool | None = PydanticUndefined, init_var: bool | None = PydanticUndefined, kw_only: bool | None = PydanticUndefined, pattern: str | Pattern[str] | None = PydanticUndefined, strict: bool | None = PydanticUndefined, coerce_numbers_to_str: bool | None = PydanticUndefined, gt: annotated_types.SupportsGt | None = PydanticUndefined, ge: annotated_types.SupportsGe | None = PydanticUndefined, lt: annotated_types.SupportsLt | None = PydanticUndefined, le: annotated_types.SupportsLe | 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, fail_fast: bool | None = PydanticUndefined, **extra: Unpack[_EmptyKwargs]) → Any

Declare field that are serialized in the http querystring.

blacksmith.domain.model.params.PostBodyField(default: Any = PydanticUndefined, *, default_factory: Callable[[], Any] | Callable[[dict[str, Any]], 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, field_title_generator: Callable[[str, FieldInfo], str] | None = PydanticUndefined, description: str | None = PydanticUndefined, examples: list[Any] | None = PydanticUndefined, exclude: bool | None = PydanticUndefined, discriminator: str | types.Discriminator | None = PydanticUndefined, deprecated: Deprecated | str | bool | None = PydanticUndefined, json_schema_extra: JsonDict | Callable[[JsonDict], None] | None = {'location': 'body'}, frozen: bool | None = PydanticUndefined, validate_default: bool | None = PydanticUndefined, repr: bool = PydanticUndefined, init: bool | None = PydanticUndefined, init_var: bool | None = PydanticUndefined, kw_only: bool | None = PydanticUndefined, pattern: str | Pattern[str] | None = PydanticUndefined, strict: bool | None = PydanticUndefined, coerce_numbers_to_str: bool | None = PydanticUndefined, gt: annotated_types.SupportsGt | None = PydanticUndefined, ge: annotated_types.SupportsGe | None = PydanticUndefined, lt: annotated_types.SupportsLt | None = PydanticUndefined, le: annotated_types.SupportsLe | 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, fail_fast: bool | None = 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].

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

class blacksmith.domain.model.params.Metadata(count: int, total_count: int | None, links: dict[str | None, dict[str, str]])

Metadata of a collection response.

count: int

total_count: int | None

links: dict[str | None, dict[str, str]]

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

Signature of the collection parser.

resp: HTTPResponse

abstract property meta: 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: HTTPResponse)

Handle the rest collection metadata parser.

Deserialize how a collection is wrapped.

total_count_header: str = 'Total-Count'

property meta: 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[Any | None]

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: Ok[HTTPResponse] | Err[HTTPError], response_schema: type[Response] | None, method: Literal['HEAD', 'GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS'], path: str, name: str, client_name: str, error_parser: AbstractErrorParser[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: dict[str, Any] | None

Return the raw json response.

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

as_result() → Ok[TResponse] | Err[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() → Ok[TResponse | None] | Err[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() → TResponse

Return the parsed response.

Raises:

NoResponseSchemaException – if there are no response schema set.

unwrap_err() → TError_co

Return the response error.

unwrap_or(default: TResponse) → 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[[TError_co], TResponse]) → 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]) → 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) → 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) → TError_co

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

map(op: Callable[[TResponse], U]) → Ok[U] | Err[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: U, op: Callable[[TResponse], U]) → 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[[], U], op: Callable[[TResponse], U]) → 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[[HTTPError], F]) → Ok[TResponse] | Err[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[[TResponse], Ok[U] | Err[HTTPError]]) → Ok[U] | Err[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[[HTTPError], Ok[TResponse] | Err[F]]) → Ok[TResponse] | Err[F]

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

Raises:

NoResponseSchemaException – if there are no response schema set.

inspect(op: Callable[[TResponse], Any]) → Ok[TResponse] | Err[TError_co]

Call op with the contained value if Ok and return the original result.

inspect_err(op: Callable[[TError_co], Any]) → Ok[TResponse] | Err[TError_co]

Call op with the contained error if Ok and return the original result.

class blacksmith.domain.model.params.CollectionIterator(response: HTTPResponse, response_schema: type[Response] | None, collection_parser: type[AbstractCollectionParser])

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

response: AbstractCollectionParser

property meta: 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.Literal['HEAD', 'GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS'], url_pattern: str, path: dict[str, str | int | float | bool] = <factory>, querystring: dict[str, str | int | float | bool | list[str | int | float | bool]] = <factory>, headers: dict[str, str] = <factory>, body: str | bytes | ~collections.abc.Iterable[bytes] | ~collections.abc.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, str | int | float | bool]

querystring: dict[str, str | int | float | bool | list[str | int | float | bool]]

headers: dict[str, str]

body: 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: Any | None)

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: Any | None

Json Body of the response.

property links: dict[str | None, dict[str, str]]