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.

blacksmith.domain.model.params.AttachmentField(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': 'attachment'}, 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.Attachment(*, filename: str, content: bytes, content_type: str | None = None, headers: ~collections.abc.Mapping[str, str] = <factory>)¶

Use to add an attachement to a multipart http query.

filename: str¶: The name of the file.

content: bytes¶: The content of the file.

content_type: str | None¶: The content type of the field. Optional, it will be guessed from the filename.

headers: Mapping[str, str]¶: The content type of the field. Optional, it will be guessed from the filename.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class blacksmith.domain.model.params.Request¶

Request Params Model.

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

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

blacksmith.domain.model.params.Response¶: Response Model.

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[BaseModel] | 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[BaseModel] | 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¶

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] | Mapping[str, str | int | float | bool] = ''¶

attachments: dict[str, tuple[str, bytes, str | None, Mapping[str, str]]] | None = None¶

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