Blacksmith API

blacksmith.scan(*modules: str) → None

Collect all resources to fillout the registry.

Basically, it import modules registered using blacksmith.register().

Raises

• TypeError – malformed module name

• ModuleNotFoundError – unknown package name

• AttributeError – argument is a module, not a package.

blacksmith.register(client_name: str, resource: str, service: str, version: Optional[str], path: Optional[str] = None, contract: Optional[Mapping[Literal['HEAD', 'GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS'], Tuple[Any, Optional[Type[blacksmith.domain.model.params.Response]]]]] = None, collection_path: Optional[str] = None, collection_contract: Optional[Mapping[Literal['HEAD', 'GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS'], Tuple[Any, Optional[Type[blacksmith.domain.model.params.Response]]]]] = None, collection_parser: Optional[Type[blacksmith.domain.model.params.AbstractCollectionParser]] = None) → None

Register a resource in a client in the default registry.

See blacksmith.domain.registry.Registry.register() for the signature.

class blacksmith.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.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.

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

Usage docs: https://docs.pydantic.dev/2.5/concepts/fields

Create a field for objects that can be configured.

Used to provide extra information about a field, either for the model schema or complex validation. Some arguments apply only to number fields (int, float, Decimal) and some apply only to str.

Note:

• Any _Unset objects will be replaced by the corresponding value defined in the _DefaultValues dictionary. If a key for the _Unset object is not found in the _DefaultValues dictionary, it will default to None

Args:

default: Default value if the field is not set. default_factory: A callable to generate the default value, such as utcnow(). alias: An alternative name for the attribute. alias_priority: Priority of the alias. This affects whether an alias generator is used. validation_alias: ‘Whitelist’ validation step. The field will be the single one allowed by the alias or set of

aliases defined.

serialization_alias: ‘Blacklist’ validation step. The vanilla field will be the single one of the alias’ or set

of aliases’ fields and all the other fields will be ignored at serialization time.

title: Human-readable title. description: Human-readable description. examples: Example values for this field. exclude: Whether to exclude the field from the model serialization. discriminator: Field name or Discriminator for discriminating the type in a tagged union. json_schema_extra: Any additional JSON schema data for the schema property. frozen: Whether the field is frozen. validate_default: Run validation that isn’t only checking existence of defaults. This can be set to True or False. If not set, it defaults to None. repr: A boolean indicating whether to include the field in the __repr__ output. init_var: Whether the field should be included in the constructor of the dataclass. kw_only: Whether the field should be a keyword-only argument in the constructor of the dataclass. strict: If True, strict validation is applied to the field.

See [Strict Mode](../concepts/strict_mode.md) for details.

gt: Greater than. If set, value must be greater than this. Only applicable to numbers. ge: Greater than or equal. If set, value must be greater than or equal to this. Only applicable to numbers. lt: Less than. If set, value must be less than this. Only applicable to numbers. le: Less than or equal. If set, value must be less than or equal to this. Only applicable to numbers. multiple_of: Value must be a multiple of this. Only applicable to numbers. min_length: Minimum length for strings. max_length: Maximum length for strings. pattern: Pattern for strings. allow_inf_nan: Allow inf, -inf, nan. Only applicable to numbers. max_digits: Maximum number of allow digits for strings. decimal_places: Maximum number of decimal places allowed for numbers. union_mode: The strategy to apply when validating a union. Can be smart (the default), or left_to_right.

See [Union Mode](standard_library_types.md#union-mode) for details.

extra: Include extra fields used by the JSON schema.

!!! warning Deprecated

The extra kwargs is deprecated. Use json_schema_extra instead.

Returns:

A new [FieldInfo][pydantic.fields.FieldInfo], the return annotation is Any so Field can be used on

type annotated fields without causing a typing error.

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

Usage docs: https://docs.pydantic.dev/2.5/concepts/fields

Create a field for objects that can be configured.

Used to provide extra information about a field, either for the model schema or complex validation. Some arguments apply only to number fields (int, float, Decimal) and some apply only to str.

Note:

• Any _Unset objects will be replaced by the corresponding value defined in the _DefaultValues dictionary. If a key for the _Unset object is not found in the _DefaultValues dictionary, it will default to None

Args:

default: Default value if the field is not set. default_factory: A callable to generate the default value, such as utcnow(). alias: An alternative name for the attribute. alias_priority: Priority of the alias. This affects whether an alias generator is used. validation_alias: ‘Whitelist’ validation step. The field will be the single one allowed by the alias or set of

aliases defined.

serialization_alias: ‘Blacklist’ validation step. The vanilla field will be the single one of the alias’ or set

of aliases’ fields and all the other fields will be ignored at serialization time.

title: Human-readable title. description: Human-readable description. examples: Example values for this field. exclude: Whether to exclude the field from the model serialization. discriminator: Field name or Discriminator for discriminating the type in a tagged union. json_schema_extra: Any additional JSON schema data for the schema property. frozen: Whether the field is frozen. validate_default: Run validation that isn’t only checking existence of defaults. This can be set to True or False. If not set, it defaults to None. repr: A boolean indicating whether to include the field in the __repr__ output. init_var: Whether the field should be included in the constructor of the dataclass. kw_only: Whether the field should be a keyword-only argument in the constructor of the dataclass. strict: If True, strict validation is applied to the field.

See [Strict Mode](../concepts/strict_mode.md) for details.

gt: Greater than. If set, value must be greater than this. Only applicable to numbers. ge: Greater than or equal. If set, value must be greater than or equal to this. Only applicable to numbers. lt: Less than. If set, value must be less than this. Only applicable to numbers. le: Less than or equal. If set, value must be less than or equal to this. Only applicable to numbers. multiple_of: Value must be a multiple of this. Only applicable to numbers. min_length: Minimum length for strings. max_length: Maximum length for strings. pattern: Pattern for strings. allow_inf_nan: Allow inf, -inf, nan. Only applicable to numbers. max_digits: Maximum number of allow digits for strings. decimal_places: Maximum number of decimal places allowed for numbers. union_mode: The strategy to apply when validating a union. Can be smart (the default), or left_to_right.

See [Union Mode](standard_library_types.md#union-mode) for details.

extra: Include extra fields used by the JSON schema.

!!! warning Deprecated

The extra kwargs is deprecated. Use json_schema_extra instead.

Returns:

A new [FieldInfo][pydantic.fields.FieldInfo], the return annotation is Any so Field can be used on

type annotated fields without causing a typing error.

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

Usage docs: https://docs.pydantic.dev/2.5/concepts/fields

Create a field for objects that can be configured.

Used to provide extra information about a field, either for the model schema or complex validation. Some arguments apply only to number fields (int, float, Decimal) and some apply only to str.

Note:

• Any _Unset objects will be replaced by the corresponding value defined in the _DefaultValues dictionary. If a key for the _Unset object is not found in the _DefaultValues dictionary, it will default to None

Args:

default: Default value if the field is not set. default_factory: A callable to generate the default value, such as utcnow(). alias: An alternative name for the attribute. alias_priority: Priority of the alias. This affects whether an alias generator is used. validation_alias: ‘Whitelist’ validation step. The field will be the single one allowed by the alias or set of

aliases defined.

serialization_alias: ‘Blacklist’ validation step. The vanilla field will be the single one of the alias’ or set

of aliases’ fields and all the other fields will be ignored at serialization time.

title: Human-readable title. description: Human-readable description. examples: Example values for this field. exclude: Whether to exclude the field from the model serialization. discriminator: Field name or Discriminator for discriminating the type in a tagged union. json_schema_extra: Any additional JSON schema data for the schema property. frozen: Whether the field is frozen. validate_default: Run validation that isn’t only checking existence of defaults. This can be set to True or False. If not set, it defaults to None. repr: A boolean indicating whether to include the field in the __repr__ output. init_var: Whether the field should be included in the constructor of the dataclass. kw_only: Whether the field should be a keyword-only argument in the constructor of the dataclass. strict: If True, strict validation is applied to the field.

See [Strict Mode](../concepts/strict_mode.md) for details.

gt: Greater than. If set, value must be greater than this. Only applicable to numbers. ge: Greater than or equal. If set, value must be greater than or equal to this. Only applicable to numbers. lt: Less than. If set, value must be less than this. Only applicable to numbers. le: Less than or equal. If set, value must be less than or equal to this. Only applicable to numbers. multiple_of: Value must be a multiple of this. Only applicable to numbers. min_length: Minimum length for strings. max_length: Maximum length for strings. pattern: Pattern for strings. allow_inf_nan: Allow inf, -inf, nan. Only applicable to numbers. max_digits: Maximum number of allow digits for strings. decimal_places: Maximum number of decimal places allowed for numbers. union_mode: The strategy to apply when validating a union. Can be smart (the default), or left_to_right.

See [Union Mode](standard_library_types.md#union-mode) for details.

extra: Include extra fields used by the JSON schema.

!!! warning Deprecated

The extra kwargs is deprecated. Use json_schema_extra instead.

Returns:

A new [FieldInfo][pydantic.fields.FieldInfo], the return annotation is Any so Field can be used on

type annotated fields without causing a typing error.

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

Usage docs: https://docs.pydantic.dev/2.5/concepts/fields

Create a field for objects that can be configured.

Used to provide extra information about a field, either for the model schema or complex validation. Some arguments apply only to number fields (int, float, Decimal) and some apply only to str.

Note:

• Any _Unset objects will be replaced by the corresponding value defined in the _DefaultValues dictionary. If a key for the _Unset object is not found in the _DefaultValues dictionary, it will default to None

Args:

default: Default value if the field is not set. default_factory: A callable to generate the default value, such as utcnow(). alias: An alternative name for the attribute. alias_priority: Priority of the alias. This affects whether an alias generator is used. validation_alias: ‘Whitelist’ validation step. The field will be the single one allowed by the alias or set of

aliases defined.

serialization_alias: ‘Blacklist’ validation step. The vanilla field will be the single one of the alias’ or set

of aliases’ fields and all the other fields will be ignored at serialization time.

title: Human-readable title. description: Human-readable description. examples: Example values for this field. exclude: Whether to exclude the field from the model serialization. discriminator: Field name or Discriminator for discriminating the type in a tagged union. json_schema_extra: Any additional JSON schema data for the schema property. frozen: Whether the field is frozen. validate_default: Run validation that isn’t only checking existence of defaults. This can be set to True or False. If not set, it defaults to None. repr: A boolean indicating whether to include the field in the __repr__ output. init_var: Whether the field should be included in the constructor of the dataclass. kw_only: Whether the field should be a keyword-only argument in the constructor of the dataclass. strict: If True, strict validation is applied to the field.

See [Strict Mode](../concepts/strict_mode.md) for details.

gt: Greater than. If set, value must be greater than this. Only applicable to numbers. ge: Greater than or equal. If set, value must be greater than or equal to this. Only applicable to numbers. lt: Less than. If set, value must be less than this. Only applicable to numbers. le: Less than or equal. If set, value must be less than or equal to this. Only applicable to numbers. multiple_of: Value must be a multiple of this. Only applicable to numbers. min_length: Minimum length for strings. max_length: Maximum length for strings. pattern: Pattern for strings. allow_inf_nan: Allow inf, -inf, nan. Only applicable to numbers. max_digits: Maximum number of allow digits for strings. decimal_places: Maximum number of decimal places allowed for numbers. union_mode: The strategy to apply when validating a union. Can be smart (the default), or left_to_right.

See [Union Mode](standard_library_types.md#union-mode) for details.

extra: Include extra fields used by the JSON schema.

!!! warning Deprecated

The extra kwargs is deprecated. Use json_schema_extra instead.

Returns:

A new [FieldInfo][pydantic.fields.FieldInfo], the return annotation is Any so Field can be used on

type annotated fields without causing a typing error.

class blacksmith.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.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.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.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.

resp: blacksmith.domain.model.http.HTTPResponse

exception blacksmith.HTTPError(message: str, request: HTTPRequest, response: HTTPResponse)

Represent the http error.

property status_code: int

property json: Optional[Any]

property is_client_error: bool

property is_server_error: bool

exception blacksmith.HTTPTimeoutError

Represent the http timeout error.

class blacksmith.AbstractErrorParser(*args, **kwargs)

A parser that parse the HTTPError class to become a pydantic model, represented by the generic TError_co here definabled in the SyncClientFactory and AsyncClientFactory.

blacksmith.default_error_parser(error: blacksmith.domain.exceptions.HTTPError) → blacksmith.domain.exceptions.HTTPError

class blacksmith.HTTPTimeout(read: float = 30.0, connect: float = 15.0)

Request timeout.

read: float

connect: float

class blacksmith.AsyncClientFactory(sd: blacksmith.sd._async.base.AsyncAbstractServiceDiscovery, transport: Optional[blacksmith.service._async.base.AsyncAbstractTransport] = None, registry: blacksmith.domain.registry.Registry = <blacksmith.domain.registry.Registry object>, timeout: Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]] = <blacksmith.domain.model.http.HTTPTimeout object>, proxies: Optional[Union[URL, str, Proxy, Dict[Union[URL, str], Union[None, URL, str, Proxy]]]] = None, verify_certificate: bool = False, collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser] = <class 'blacksmith.domain.model.params.CollectionParser'>, error_parser: Optional[blacksmith.domain.error.AbstractErrorParser[blacksmith.domain.error.TError_co]] = None)

Client creator, for the given configuration.

Parameters

• sd – Service Discovery instance

• transport – HTTP Client that process the call, default use blacksmith.service._async.adapters.httpx.HttpxTransport

• timeout – configure timeout, this parameter is ignored if the transport has been passed

• proxies – configure proxies, this parameter is ignored if the transport has been passed

• verify_certificate – Reject request if certificate are invalid for https

• collection_parser – use to customize the collection parser default use blacksmith.domain.model.params.CollectionParser

sd: blacksmith.sd._async.base.AsyncAbstractServiceDiscovery

registry: blacksmith.domain.registry.Registry

transport: blacksmith.service._async.base.AsyncAbstractTransport

timeout: blacksmith.domain.model.http.HTTPTimeout

collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser]

error_parser: blacksmith.domain.error.AbstractErrorParser[blacksmith.domain.error.TError_co]

middlewares: List[blacksmith.middleware._async.base.AsyncHTTPMiddleware]

add_middleware(middleware: blacksmith.middleware._async.base.AsyncHTTPMiddleware) → blacksmith.service._async.client.AsyncClientFactory[blacksmith.domain.error.TError_co]

Add a middleware to the client factory and return the client for chaining.

..note:: Clients created before the call of this method will also be

altered. The middleware stack is a reference for all clients.

async initialize() → None

class blacksmith.SyncClientFactory(sd: blacksmith.sd._sync.base.SyncAbstractServiceDiscovery, transport: Optional[blacksmith.service._sync.base.SyncAbstractTransport] = None, registry: blacksmith.domain.registry.Registry = <blacksmith.domain.registry.Registry object>, timeout: Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]] = <blacksmith.domain.model.http.HTTPTimeout object>, proxies: Optional[Union[URL, str, Proxy, Dict[Union[URL, str], Union[None, URL, str, Proxy]]]] = None, verify_certificate: bool = False, collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser] = <class 'blacksmith.domain.model.params.CollectionParser'>, error_parser: Optional[blacksmith.domain.error.AbstractErrorParser[blacksmith.domain.error.TError_co]] = None)

Client creator, for the given configuration.

Parameters

• sd – Service Discovery instance

• transport – HTTP Client that process the call, default use blacksmith.service._async.adapters.httpx.HttpxTransport

• timeout – configure timeout, this parameter is ignored if the transport has been passed

• proxies – configure proxies, this parameter is ignored if the transport has been passed

• verify_certificate – Reject request if certificate are invalid for https

• collection_parser – use to customize the collection parser default use blacksmith.domain.model.params.CollectionParser

sd: blacksmith.sd._sync.base.SyncAbstractServiceDiscovery

registry: blacksmith.domain.registry.Registry

transport: blacksmith.service._sync.base.SyncAbstractTransport

timeout: blacksmith.domain.model.http.HTTPTimeout

collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser]

error_parser: blacksmith.domain.error.AbstractErrorParser[blacksmith.domain.error.TError_co]

middlewares: List[blacksmith.middleware._sync.base.SyncHTTPMiddleware]

add_middleware(middleware: blacksmith.middleware._sync.base.SyncHTTPMiddleware) → blacksmith.service._sync.client.SyncClientFactory[blacksmith.domain.error.TError_co]

Add a middleware to the client factory and return the client for chaining.

..note:: Clients created before the call of this method will also be

altered. The middleware stack is a reference for all clients.

initialize() → None

class blacksmith.AsyncClient(name: str, endpoint: str, resources: Mapping[str, blacksmith.domain.registry.ApiRoutes], transport: blacksmith.service._async.base.AsyncAbstractTransport, timeout: blacksmith.domain.model.http.HTTPTimeout, collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser], middlewares: List[blacksmith.middleware._async.base.AsyncHTTPMiddleware], error_parser: blacksmith.domain.error.AbstractErrorParser[blacksmith.domain.error.TError_co])

Client representation for the client name.

A client will have dymanic property, based on the registered resources.

name: str

endpoint: str

resources: Mapping[str, blacksmith.domain.registry.ApiRoutes]

transport: blacksmith.service._async.base.AsyncAbstractTransport

timeout: blacksmith.domain.model.http.HTTPTimeout

collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser]

middlewares: List[blacksmith.middleware._async.base.AsyncHTTPMiddleware]

add_middleware(middleware: blacksmith.middleware._async.base.AsyncHTTPMiddleware) → blacksmith.service._async.client.AsyncClient[blacksmith.domain.error.TError_co]

class blacksmith.SyncClient(name: str, endpoint: str, resources: Mapping[str, blacksmith.domain.registry.ApiRoutes], transport: blacksmith.service._sync.base.SyncAbstractTransport, timeout: blacksmith.domain.model.http.HTTPTimeout, collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser], middlewares: List[blacksmith.middleware._sync.base.SyncHTTPMiddleware], error_parser: blacksmith.domain.error.AbstractErrorParser[blacksmith.domain.error.TError_co])

Client representation for the client name.

A client will have dymanic property, based on the registered resources.

name: str

endpoint: str

resources: Mapping[str, blacksmith.domain.registry.ApiRoutes]

transport: blacksmith.service._sync.base.SyncAbstractTransport

timeout: blacksmith.domain.model.http.HTTPTimeout

collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser]

middlewares: List[blacksmith.middleware._sync.base.SyncHTTPMiddleware]

add_middleware(middleware: blacksmith.middleware._sync.base.SyncHTTPMiddleware) → blacksmith.service._sync.client.SyncClient[blacksmith.domain.error.TError_co]

class blacksmith.AsyncRouteProxy(client_name: str, name: str, endpoint: str, routes: blacksmith.domain.registry.ApiRoutes, transport: blacksmith.service._async.base.AsyncAbstractTransport, timeout: blacksmith.domain.model.http.HTTPTimeout, collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser], error_parser: blacksmith.domain.error.AbstractErrorParser[blacksmith.domain.error.TError_co], middlewares: List[blacksmith.middleware._async.base.AsyncHTTPMiddleware])

Proxy from resource to its associate routes.

client_name: str

name: str

endpoint: str

routes: blacksmith.domain.registry.ApiRoutes

transport: blacksmith.service._async.base.AsyncAbstractTransport

timeout: blacksmith.domain.model.http.HTTPTimeout

collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser]

error_parser: blacksmith.domain.error.AbstractErrorParser[blacksmith.domain.error.TError_co]

middlewares: List[blacksmith.middleware._async.base.AsyncHTTPMiddleware]

async collection_head(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http HEAD query on the collection_path.

async collection_get(params: Union[blacksmith.domain.model.params.Request, None, Dict[Any, Any]] = None, timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → Union[result.result.Ok[blacksmith.domain.model.params.CollectionIterator[blacksmith.domain.model.params.TCollectionResponse]], result.result.Err[blacksmith.domain.error.TError_co]]

Retrieve a collection of resources.

It perform an http GET query on the collection_path.

The collection is return in as an iterator, and models ares validated one by one using the TCollectionResponse schema which descrine one item of the collection.

Important

This method is the only method that behave as an iterator. You can update the way collection are deserialize for a whole client, by passing a blacksmith.AbstractCollectionParser on the blacksmith.AsyncClientFactory ( or blacksmith.SyncClientFactory for the synchronous version).

async collection_post(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http POST query on the collection_path.

async collection_put(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http PUT query on the collection_path.

async collection_patch(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http PATCH query on the collection_path.

async collection_delete(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http DELETE query on the collection_path.

async collection_options(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http OPTIONS query on the collection_path.

async head(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http HEAD query on the path.

async get(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http GET query on the path.

async post(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http POST query on the path.

async put(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http PUT query on the path.

async patch(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http PATCH query on the path.

async delete(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http DELETE query on the path.

async options(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http OPTIONS query on the path.

class blacksmith.SyncRouteProxy(client_name: str, name: str, endpoint: str, routes: blacksmith.domain.registry.ApiRoutes, transport: blacksmith.service._sync.base.SyncAbstractTransport, timeout: blacksmith.domain.model.http.HTTPTimeout, collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser], error_parser: blacksmith.domain.error.AbstractErrorParser[blacksmith.domain.error.TError_co], middlewares: List[blacksmith.middleware._sync.base.SyncHTTPMiddleware])

Proxy from resource to its associate routes.

client_name: str

name: str

endpoint: str

routes: blacksmith.domain.registry.ApiRoutes

transport: blacksmith.service._sync.base.SyncAbstractTransport

timeout: blacksmith.domain.model.http.HTTPTimeout

collection_parser: Type[blacksmith.domain.model.params.AbstractCollectionParser]

error_parser: blacksmith.domain.error.AbstractErrorParser[blacksmith.domain.error.TError_co]

middlewares: List[blacksmith.middleware._sync.base.SyncHTTPMiddleware]

collection_head(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http HEAD query on the collection_path.

collection_get(params: Union[blacksmith.domain.model.params.Request, None, Dict[Any, Any]] = None, timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → Union[result.result.Ok[blacksmith.domain.model.params.CollectionIterator[blacksmith.domain.model.params.TCollectionResponse]], result.result.Err[blacksmith.domain.error.TError_co]]

Retrieve a collection of resources.

It perform an http GET query on the collection_path.

The collection is return in as an iterator, and models ares validated one by one using the TCollectionResponse schema which descrine one item of the collection.

Important

This method is the only method that behave as an iterator. You can update the way collection are deserialize for a whole client, by passing a blacksmith.AbstractCollectionParser on the blacksmith.AsyncClientFactory ( or blacksmith.SyncClientFactory for the synchronous version).

collection_post(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http POST query on the collection_path.

collection_put(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http PUT query on the collection_path.

collection_patch(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http PATCH query on the collection_path.

collection_delete(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http DELETE query on the collection_path.

collection_options(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http OPTIONS query on the collection_path.

head(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http HEAD query on the path.

get(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http GET query on the path.

post(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http POST query on the path.

put(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http PUT query on the path.

patch(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http PATCH query on the path.

delete(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http DELETE query on the path.

options(params: Union[blacksmith.domain.model.params.Request, Dict[Any, Any]], timeout: Optional[Union[blacksmith.domain.model.http.HTTPTimeout, float, Tuple[float, float]]] = None) → blacksmith.domain.model.params.ResponseBox[blacksmith.domain.model.params.TResponse, blacksmith.domain.error.TError_co]

Use to perform an http OPTIONS query on the path.

class blacksmith.AsyncAbstractServiceDiscovery

Define the Service Discovery interface.

abstract async get_endpoint(service: str, version: Optional[str]) → str

Get the endpoint of a service.

class blacksmith.SyncAbstractServiceDiscovery

Define the Service Discovery interface.

abstract get_endpoint(service: str, version: Optional[str]) → str

Get the endpoint of a service.

class blacksmith.AsyncConsulDiscovery(addr: str = 'http://consul:8500/v1', service_name_fmt: str = '{service}-{version}', service_url_fmt: str = 'http://{address}:{port}/{version}', unversioned_service_name_fmt: str = '{service}', unversioned_service_url_fmt: str = 'http://{address}:{port}', consul_token: str = '', _client_factory: Callable[[str, str], blacksmith.service._async.client.AsyncClientFactory[Any]] = <function blacksmith_cli>)

A discovery instance based on a Consul server.

Parameters

• service_name_fmt – pattern for name of versionned service

• service_url_fmt – pattern for url of versionned service

• unversioned_service_name_fmt – pattern for name of unversioned service

• unversioned_service_url_fmt – pattern for url of unversioned service

service_name_fmt: str

service_url_fmt: str

unversioned_service_name_fmt: str

unversioned_service_url_fmt: str

format_service_name(service: str, version: Optional[str]) → str

Build the service name to send to consul.

format_endoint(version: Optional[str], address: str, port: int) → str

Build the rest api endpoint from consul response.

async resolve(service: str, version: Optional[str]) → blacksmith.sd._async.adapters.consul.Service

Get the Service from the consul registry.

If many instances host the service, the host is choosen randomly.

async get_endpoint(service: str, version: Optional[str]) → str

Get the endpoint from the consul registry

If many instances host the service, the host is choosen randomly.

class blacksmith.SyncConsulDiscovery(addr: str = 'http://consul:8500/v1', service_name_fmt: str = '{service}-{version}', service_url_fmt: str = 'http://{address}:{port}/{version}', unversioned_service_name_fmt: str = '{service}', unversioned_service_url_fmt: str = 'http://{address}:{port}', consul_token: str = '', _client_factory: Callable[[str, str], blacksmith.service._sync.client.SyncClientFactory[Any]] = <function blacksmith_cli>)

A discovery instance based on a Consul server.

Parameters

• service_name_fmt – pattern for name of versionned service

• service_url_fmt – pattern for url of versionned service

• unversioned_service_name_fmt – pattern for name of unversioned service

• unversioned_service_url_fmt – pattern for url of unversioned service

service_name_fmt: str

service_url_fmt: str

unversioned_service_name_fmt: str

unversioned_service_url_fmt: str

format_service_name(service: str, version: Optional[str]) → str

Build the service name to send to consul.

format_endoint(version: Optional[str], address: str, port: int) → str

Build the rest api endpoint from consul response.

resolve(service: str, version: Optional[str]) → blacksmith.sd._sync.adapters.consul.Service

Get the Service from the consul registry.

If many instances host the service, the host is choosen randomly.

get_endpoint(service: str, version: Optional[str]) → str

Get the endpoint from the consul registry

If many instances host the service, the host is choosen randomly.

class blacksmith.AsyncRouterDiscovery(service_url_fmt: str = 'http://router/{service}-{version}/{version}', unversioned_service_url_fmt: str = 'http://router/{service}')

Router that implement a Server-Side Service discovery.

This implementation never raise blacksmith.domain.exceptions.UnregisteredServiceException when service are unknown, because it only passe very request to the router server that is connected to the service registry.

Note

Given pattern in parameter have to match the format of the router server.

Parameters

• service_url_fmt – A pattern used to create endpoint of versionned services.

• unversioned_service_url_fmt – A pattern used to create endpoint of unversionned services.

service_url_fmt: str

unversioned_service_url_fmt: str

async get_endpoint(service: str, version: Optional[str]) → str

Create and return the endpoint using the given parameters service_url_fmt or unversioned_service_url_fmt if version is None.

class blacksmith.SyncRouterDiscovery(service_url_fmt: str = 'http://router/{service}-{version}/{version}', unversioned_service_url_fmt: str = 'http://router/{service}')

Router that implement a Server-Side Service discovery.

This implementation never raise blacksmith.domain.exceptions.UnregisteredServiceException when service are unknown, because it only passe very request to the router server that is connected to the service registry.

Note

Given pattern in parameter have to match the format of the router server.

Parameters

• service_url_fmt – A pattern used to create endpoint of versionned services.

• unversioned_service_url_fmt – A pattern used to create endpoint of unversionned services.

service_url_fmt: str

unversioned_service_url_fmt: str

get_endpoint(service: str, version: Optional[str]) → str

Create and return the endpoint using the given parameters service_url_fmt or unversioned_service_url_fmt if version is None.

class blacksmith.AsyncStaticDiscovery(endpoints: Mapping[Tuple[str, Optional[str]], str])

A discovery instance based on a static dictionary.

endpoints: Mapping[Tuple[str, Optional[str]], str]

async get_endpoint(service: str, version: Optional[str]) → str

Retrieve endpoint using the given parameters from endpoints.

class blacksmith.SyncStaticDiscovery(endpoints: Mapping[Tuple[str, Optional[str]], str])

A discovery instance based on a static dictionary.

endpoints: Mapping[Tuple[str, Optional[str]], str]

get_endpoint(service: str, version: Optional[str]) → str

Retrieve endpoint using the given parameters from endpoints.

class blacksmith.AsyncMiddleware(*args, **kwargs)

Signature of the middleware for the async version.

class blacksmith.SyncMiddleware(*args, **kwargs)

Signature of the middleware for the sync version.

class blacksmith.AsyncHTTPMiddleware

Inject data in http query on every requests.

async initialize() → None

Asynchronous initialization of a middleware.

For instance, used to initialize connection to storage backend.

class blacksmith.SyncHTTPMiddleware

Inject data in http query on every requests.

initialize() → None

Asynchronous initialization of a middleware.

For instance, used to initialize connection to storage backend.

class blacksmith.AsyncHTTPAddHeadersMiddleware(headers: Dict[str, str])

Generic middleware that inject HTTP headers.

Params

headers to inject in HTTP requests.

headers: Dict[str, str]

class blacksmith.SyncHTTPAddHeadersMiddleware(headers: Dict[str, str])

Generic middleware that inject HTTP headers.

Params

headers to inject in HTTP requests.

headers: Dict[str, str]

class blacksmith.AsyncHTTPAuthorizationMiddleware(scheme: str, value: str)

Authentication Mechanism based on the header Authorization.

Parameters

• scheme – the scheme of the mechanism.

• value – the value that authenticate the user using the scheme.

class blacksmith.SyncHTTPAuthorizationMiddleware(scheme: str, value: str)

Authentication Mechanism based on the header Authorization.

Parameters

• scheme – the scheme of the mechanism.

• value – the value that authenticate the user using the scheme.

class blacksmith.AsyncHTTPBearerMiddleware(bearer_token: str)

Authentication Mechanism based on the header Authorization with the Bearer scheme.

Parameters

value – value of the bearer token.

class blacksmith.SyncHTTPBearerMiddleware(bearer_token: str)

Authentication Mechanism based on the header Authorization with the Bearer scheme.

Parameters

value – value of the bearer token.

class blacksmith.AsyncCircuitBreakerMiddleware(threshold: int = 5, ttl: float = 30, listeners: Optional[Iterable[Callable[[str, Literal['circuit_breaker_created', 'state_changed', 'failed', 'recovered'], purgatory.domain.messages.base.Event], None]]] = None, uow: Optional[purgatory.service._async.unit_of_work.AsyncAbstractUnitOfWork] = None, metrics: Optional[blacksmith.domain.model.middleware.prometheus.PrometheusMetrics] = None)

Prevent cascading failure.

The circuit breaker is based on purgatory, the middleware create one circuit breaker per client_name. The parameters are forwarded to all the clients. This middleware does not give the possibility to adapt a threshold or the time the circuit is opened per clients.

async initialize() → None

Asynchronous initialization of a middleware.

For instance, used to initialize connection to storage backend.

class blacksmith.SyncCircuitBreakerMiddleware(threshold: int = 5, ttl: float = 30, listeners: Optional[Iterable[Callable[[str, Literal['circuit_breaker_created', 'state_changed', 'failed', 'recovered'], purgatory.domain.messages.base.Event], None]]] = None, uow: Optional[purgatory.service._sync.unit_of_work.SyncAbstractUnitOfWork] = None, metrics: Optional[blacksmith.domain.model.middleware.prometheus.PrometheusMetrics] = None)

Prevent cascading failure.

The circuit breaker is based on purgatory, the middleware create one circuit breaker per client_name. The parameters are forwarded to all the clients. This middleware does not give the possibility to adapt a threshold or the time the circuit is opened per clients.

initialize() → None

Asynchronous initialization of a middleware.

For instance, used to initialize connection to storage backend.

class blacksmith.PrometheusMetrics(buckets: Optional[List[float]] = None, hit_cache_buckets: Optional[List[float]] = None, registry: Optional[Any] = None)

class blacksmith.AsyncPrometheusMiddleware(metrics: Optional[blacksmith.domain.model.middleware.prometheus.PrometheusMetrics] = None)

Collect the api calls made in a prometheus registry.

It expose a blacksmith_info Gauge to get the blacksmith version, as a label, and a blacksmith_request_latency_seconds_count Counter to get the number of http requests made. The counter blacksmith_request_latency_seconds_count as client_name, method, path and status_code labels.

Note

the service_name and service version is redundant with the client_name, so they are not exposed as labels. By the way, you may have multiple client_name for 1 service name/version.

metrics: blacksmith.domain.model.middleware.prometheus.PrometheusMetrics

class blacksmith.SyncPrometheusMiddleware(metrics: Optional[blacksmith.domain.model.middleware.prometheus.PrometheusMetrics] = None)

Collect the api calls made in a prometheus registry.

It expose a blacksmith_info Gauge to get the blacksmith version, as a label, and a blacksmith_request_latency_seconds_count Counter to get the number of http requests made. The counter blacksmith_request_latency_seconds_count as client_name, method, path and status_code labels.

Note

the service_name and service version is redundant with the client_name, so they are not exposed as labels. By the way, you may have multiple client_name for 1 service name/version.

metrics: blacksmith.domain.model.middleware.prometheus.PrometheusMetrics

class blacksmith.AbstractCachePolicy

Define the Cache Policy

abstract handle_request(req: blacksmith.domain.model.http.HTTPRequest, client_name: str, path: str) → bool

A function to decide if the http request is cachable.

abstract get_vary_key(client_name: str, path: str, request: blacksmith.domain.model.http.HTTPRequest) → str

Create a caching key for the vary part.

abstract get_response_cache_key(client_name: str, path: str, req: blacksmith.domain.model.http.HTTPRequest, vary: List[str]) → str

Create a caching key for the http response.

abstract get_cache_info_for_response(client_name: str, path: str, req: blacksmith.domain.model.http.HTTPRequest, resp: blacksmith.domain.model.http.HTTPResponse) → Tuple[int, str, List[str]]

Return caching info. Tuple (ttl in seconds, vary key, vary list).

class blacksmith.AbstractSerializer

abstract static loads(s: str) → Any

Load a string to an object

abstract static dumps(obj: Any) → str

Get a value from redis

class blacksmith.CacheControlPolicy(sep: str = '$')

Initialize the caching using Cache-Control http headers. Also consume the Vary response header to cache response per Vary response headers per request.

Parameters

sep – Separator used in cache key MUST NOT BE USED in client name.

handle_request(req: blacksmith.domain.model.http.HTTPRequest, client_name: str, path: str) → bool

A function to decide if the http request is cachable.

get_vary_key(client_name: str, path: str, request: blacksmith.domain.model.http.HTTPRequest) → str

Create a caching key for the vary part.

get_response_cache_key(client_name: str, path: str, req: blacksmith.domain.model.http.HTTPRequest, vary: List[str]) → str

Create a caching key for the http response.

get_cache_info_for_response(client_name: str, path: str, req: blacksmith.domain.model.http.HTTPRequest, resp: blacksmith.domain.model.http.HTTPResponse) → Tuple[int, str, List[str]]

Return caching info. Tuple (ttl in seconds, vary key, vary list).

class blacksmith.JsonSerializer

static loads(s: str) → Any

Load a string to an object

static dumps(obj: Any) → str

Get a value from redis

class blacksmith.AsyncAbstractCache

Abstract Redis Client.

abstract async initialize() → None

Initialize the cache

abstract async get(key: str) → Optional[str]

Get a value from redis

abstract async set(key: str, val: str, ex: datetime.timedelta) → None

Get a value from redis

class blacksmith.AsyncHTTPCacheMiddleware(cache: blacksmith.middleware._async.http_cache.AsyncAbstractCache, metrics: Optional[blacksmith.domain.model.middleware.prometheus.PrometheusMetrics] = None, policy: blacksmith.domain.model.middleware.http_cache.AbstractCachePolicy = <blacksmith.domain.model.middleware.http_cache.CacheControlPolicy object>, serializer: Type[blacksmith.domain.model.middleware.http_cache.AbstractSerializer] = <class 'blacksmith.domain.model.middleware.http_cache.JsonSerializer'>)

Http Cache Middleware based on Cache-Control and redis.

async initialize() → None

Asynchronous initialization of a middleware.

For instance, used to initialize connection to storage backend.

async cache_response(client_name: str, path: str, req: blacksmith.domain.model.http.HTTPRequest, resp: blacksmith.domain.model.http.HTTPResponse) → bool

async get_from_cache(client_name: str, path: str, req: blacksmith.domain.model.http.HTTPRequest) → Optional[blacksmith.domain.model.http.HTTPResponse]

observe_cache_hit(client_name: str, method: str, path: str, status_code: int, latency: float) → None

inc_cache_miss(client_name: str, cachable_state: Literal['uncachable_request', 'uncachable_response', 'cached'], method: Literal['HEAD', 'GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS'], path: str, status_code: int) → None

class blacksmith.SyncHTTPCacheMiddleware(cache: blacksmith.middleware._sync.http_cache.SyncAbstractCache, metrics: Optional[blacksmith.domain.model.middleware.prometheus.PrometheusMetrics] = None, policy: blacksmith.domain.model.middleware.http_cache.AbstractCachePolicy = <blacksmith.domain.model.middleware.http_cache.CacheControlPolicy object>, serializer: Type[blacksmith.domain.model.middleware.http_cache.AbstractSerializer] = <class 'blacksmith.domain.model.middleware.http_cache.JsonSerializer'>)

Http Cache Middleware based on Cache-Control and redis.

initialize() → None

Asynchronous initialization of a middleware.

For instance, used to initialize connection to storage backend.

cache_response(client_name: str, path: str, req: blacksmith.domain.model.http.HTTPRequest, resp: blacksmith.domain.model.http.HTTPResponse) → bool

get_from_cache(client_name: str, path: str, req: blacksmith.domain.model.http.HTTPRequest) → Optional[blacksmith.domain.model.http.HTTPResponse]

observe_cache_hit(client_name: str, method: str, path: str, status_code: int, latency: float) → None

inc_cache_miss(client_name: str, cachable_state: Literal['uncachable_request', 'uncachable_response', 'cached'], method: Literal['HEAD', 'GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS'], path: str, status_code: int) → None

class blacksmith.AbstractTraceContext(name: str, kind: str = 'SERVER')

Interface of the trace context for the middleware.

See examples with starlette-zipking for an implementation.

abstract classmethod make_headers() → Dict[str, str]

Build headers for the sub requests.

abstract tag(key: str, value: str) → blacksmith.domain.model.middleware.zipkin.AbstractTraceContext

Tag the span

abstract annotate(value: Optional[str], ts: Optional[float] = None) → blacksmith.domain.model.middleware.zipkin.AbstractTraceContext

Annotate the span

class blacksmith.AsyncZipkinMiddleware(trace: Type[blacksmith.middleware._async.zipkin.AbstractTraceContext])

Zipkin Middleware based on an abstract context manager.

Parameters

trace – A deferred context manager that manage the trace span stack.

class blacksmith.AsyncAbstractTransport(verify_certificate: bool = True, proxies: Optional[Union[URL, str, Proxy, Dict[Union[URL, str], Union[None, URL, str, Proxy]]]] = None)

verify_certificate: bool

proxies: Optional[Union[URL, str, Proxy, Dict[Union[URL, str], Union[None, URL, str, Proxy]]]]

class blacksmith.SyncAbstractTransport(verify_certificate: bool = True, proxies: Optional[Union[URL, str, Proxy, Dict[Union[URL, str], Union[None, URL, str, Proxy]]]] = None)

verify_certificate: bool

proxies: Optional[Union[URL, str, Proxy, Dict[Union[URL, str], Union[None, URL, str, Proxy]]]]

class blacksmith.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

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

class blacksmith.AbstractHttpBodySerializer

Request body serializer.

abstract accept(content_type: str) → bool

Return true in case it can handle the request.

abstract serialize(body: Union[Dict[str, Any], Sequence[Any]]) → Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]]

Serialize a python simple types to a python request body.

The body received here is the extracted object from the request model.

abstract deserialize(body: bytes, encoding: Optional[str]) → Optional[Any]

Deserialize a raw http response body to a python simple types representation.

blacksmith.register_http_body_serializer(serializer: blacksmith.service.http_body_serializer.AbstractHttpBodySerializer) → None

Register a serializer to serialize some kind of request.

blacksmith.unregister_http_body_serializer(serializer: blacksmith.service.http_body_serializer.AbstractHttpBodySerializer) → None

Unregister a serializer previously added.

Usefull for testing purpose.