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[typing_extensions.Literal[HEAD, GET, POST, PUT, PATCH, DELETE, OPTIONS], Tuple[Type[blacksmith.domain.model.params.Request], Optional[Type[blacksmith.domain.model.params.Response]]]]] = None, collection_path: Optional[str] = None, collection_contract: Optional[Mapping[typing_extensions.Literal[HEAD, GET, POST, PUT, PATCH, DELETE, OPTIONS], Tuple[Type[blacksmith.domain.model.params.Request], 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()
orPostBodyField()
to declare each fields.- to_http_request(method: typing_extensions.Literal[HEAD, GET, POST, PUT, PATCH, DELETE, OPTIONS], url_pattern: str) blacksmith.domain.model.http.HTTPRequest #
Convert the request params to an http request in order to serialize the http request for the client.
- class blacksmith.Response#
Response Model.
- blacksmith.HeaderField(default: Any = PydanticUndefined, *, default_factory: Optional[Callable[[], Any]] = None, alias: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, const: Optional[bool] = None, gt: Optional[float] = None, ge: Optional[float] = None, lt: Optional[float] = None, le: Optional[float] = None, multiple_of: Optional[float] = None, allow_inf_nan: Optional[bool] = None, max_digits: Optional[int] = None, decimal_places: Optional[int] = None, min_items: Optional[int] = None, max_items: Optional[int] = None, unique_items: Optional[bool] = None, min_length: Optional[int] = None, max_length: Optional[int] = None, allow_mutation: bool = True, regex: Optional[str] = None, discriminator: Optional[str] = None, repr: bool = True, **extra: Any) Any #
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 tostr
.- Parameters
default – since this is replacing the field’s default, its first argument is used to set the default, use ellipsis (
...
) to indicate the field is requireddefault_factory – callable that will be called when a default value is needed for this field If both default and default_factory are set, an error is raised.
alias – the public name of the field
title – can be any string, used in the schema
description – can be any string, used in the schema
exclude – exclude this field while dumping. Takes same values as the
include
andexclude
arguments on the.dict
method.include – include this field while dumping. Takes same values as the
include
andexclude
arguments on the.dict
method.const – this field is required and must take it’s default value
gt – only applies to numbers, requires the field to be “greater than”. The schema will have an
exclusiveMinimum
validation keywordge – only applies to numbers, requires the field to be “greater than or equal to”. The schema will have a
minimum
validation keywordlt – only applies to numbers, requires the field to be “less than”. The schema will have an
exclusiveMaximum
validation keywordle – only applies to numbers, requires the field to be “less than or equal to”. The schema will have a
maximum
validation keywordmultiple_of – only applies to numbers, requires the field to be “a multiple of”. The schema will have a
multipleOf
validation keywordallow_inf_nan – only applies to numbers, allows the field to be NaN or infinity (+inf or -inf), which is a valid Python float. Default True, set to False for compatibility with JSON.
max_digits – only applies to Decimals, requires the field to have a maximum number of digits within the decimal. It does not include a zero before the decimal point or trailing decimal zeroes.
decimal_places – only applies to Decimals, requires the field to have at most a number of decimal places allowed. It does not include trailing decimal zeroes.
min_items – only applies to lists, requires the field to have a minimum number of elements. The schema will have a
minItems
validation keywordmax_items – only applies to lists, requires the field to have a maximum number of elements. The schema will have a
maxItems
validation keywordunique_items – only applies to lists, requires the field not to have duplicated elements. The schema will have a
uniqueItems
validation keywordmin_length – only applies to strings, requires the field to have a minimum length. The schema will have a
minLength
validation keywordmax_length – only applies to strings, requires the field to have a maximum length. The schema will have a
maxLength
validation keywordallow_mutation – a boolean which defaults to True. When False, the field raises a TypeError if the field is assigned on an instance. The BaseModel Config must set validate_assignment to True
regex – only applies to strings, requires the field match against a regular expression pattern string. The schema will have a
pattern
validation keyworddiscriminator – only useful with a (discriminated a.k.a. tagged) Union of sub models with a common field. The discriminator is the name of this common field to shorten validation and improve generated schema
repr – show this field in the representation
**extra – any additional keyword arguments will be added as is to the schema
- blacksmith.PathInfoField(default: Any = PydanticUndefined, *, default_factory: Optional[Callable[[], Any]] = None, alias: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, const: Optional[bool] = None, gt: Optional[float] = None, ge: Optional[float] = None, lt: Optional[float] = None, le: Optional[float] = None, multiple_of: Optional[float] = None, allow_inf_nan: Optional[bool] = None, max_digits: Optional[int] = None, decimal_places: Optional[int] = None, min_items: Optional[int] = None, max_items: Optional[int] = None, unique_items: Optional[bool] = None, min_length: Optional[int] = None, max_length: Optional[int] = None, allow_mutation: bool = True, regex: Optional[str] = None, discriminator: Optional[str] = None, repr: bool = True, **extra: Any) Any #
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 tostr
.- Parameters
default – since this is replacing the field’s default, its first argument is used to set the default, use ellipsis (
...
) to indicate the field is requireddefault_factory – callable that will be called when a default value is needed for this field If both default and default_factory are set, an error is raised.
alias – the public name of the field
title – can be any string, used in the schema
description – can be any string, used in the schema
exclude – exclude this field while dumping. Takes same values as the
include
andexclude
arguments on the.dict
method.include – include this field while dumping. Takes same values as the
include
andexclude
arguments on the.dict
method.const – this field is required and must take it’s default value
gt – only applies to numbers, requires the field to be “greater than”. The schema will have an
exclusiveMinimum
validation keywordge – only applies to numbers, requires the field to be “greater than or equal to”. The schema will have a
minimum
validation keywordlt – only applies to numbers, requires the field to be “less than”. The schema will have an
exclusiveMaximum
validation keywordle – only applies to numbers, requires the field to be “less than or equal to”. The schema will have a
maximum
validation keywordmultiple_of – only applies to numbers, requires the field to be “a multiple of”. The schema will have a
multipleOf
validation keywordallow_inf_nan – only applies to numbers, allows the field to be NaN or infinity (+inf or -inf), which is a valid Python float. Default True, set to False for compatibility with JSON.
max_digits – only applies to Decimals, requires the field to have a maximum number of digits within the decimal. It does not include a zero before the decimal point or trailing decimal zeroes.
decimal_places – only applies to Decimals, requires the field to have at most a number of decimal places allowed. It does not include trailing decimal zeroes.
min_items – only applies to lists, requires the field to have a minimum number of elements. The schema will have a
minItems
validation keywordmax_items – only applies to lists, requires the field to have a maximum number of elements. The schema will have a
maxItems
validation keywordunique_items – only applies to lists, requires the field not to have duplicated elements. The schema will have a
uniqueItems
validation keywordmin_length – only applies to strings, requires the field to have a minimum length. The schema will have a
minLength
validation keywordmax_length – only applies to strings, requires the field to have a maximum length. The schema will have a
maxLength
validation keywordallow_mutation – a boolean which defaults to True. When False, the field raises a TypeError if the field is assigned on an instance. The BaseModel Config must set validate_assignment to True
regex – only applies to strings, requires the field match against a regular expression pattern string. The schema will have a
pattern
validation keyworddiscriminator – only useful with a (discriminated a.k.a. tagged) Union of sub models with a common field. The discriminator is the name of this common field to shorten validation and improve generated schema
repr – show this field in the representation
**extra – any additional keyword arguments will be added as is to the schema
- blacksmith.PostBodyField(default: Any = PydanticUndefined, *, default_factory: Optional[Callable[[], Any]] = None, alias: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, const: Optional[bool] = None, gt: Optional[float] = None, ge: Optional[float] = None, lt: Optional[float] = None, le: Optional[float] = None, multiple_of: Optional[float] = None, allow_inf_nan: Optional[bool] = None, max_digits: Optional[int] = None, decimal_places: Optional[int] = None, min_items: Optional[int] = None, max_items: Optional[int] = None, unique_items: Optional[bool] = None, min_length: Optional[int] = None, max_length: Optional[int] = None, allow_mutation: bool = True, regex: Optional[str] = None, discriminator: Optional[str] = None, repr: bool = True, **extra: Any) Any #
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 tostr
.- Parameters
default – since this is replacing the field’s default, its first argument is used to set the default, use ellipsis (
...
) to indicate the field is requireddefault_factory – callable that will be called when a default value is needed for this field If both default and default_factory are set, an error is raised.
alias – the public name of the field
title – can be any string, used in the schema
description – can be any string, used in the schema
exclude – exclude this field while dumping. Takes same values as the
include
andexclude
arguments on the.dict
method.include – include this field while dumping. Takes same values as the
include
andexclude
arguments on the.dict
method.const – this field is required and must take it’s default value
gt – only applies to numbers, requires the field to be “greater than”. The schema will have an
exclusiveMinimum
validation keywordge – only applies to numbers, requires the field to be “greater than or equal to”. The schema will have a
minimum
validation keywordlt – only applies to numbers, requires the field to be “less than”. The schema will have an
exclusiveMaximum
validation keywordle – only applies to numbers, requires the field to be “less than or equal to”. The schema will have a
maximum
validation keywordmultiple_of – only applies to numbers, requires the field to be “a multiple of”. The schema will have a
multipleOf
validation keywordallow_inf_nan – only applies to numbers, allows the field to be NaN or infinity (+inf or -inf), which is a valid Python float. Default True, set to False for compatibility with JSON.
max_digits – only applies to Decimals, requires the field to have a maximum number of digits within the decimal. It does not include a zero before the decimal point or trailing decimal zeroes.
decimal_places – only applies to Decimals, requires the field to have at most a number of decimal places allowed. It does not include trailing decimal zeroes.
min_items – only applies to lists, requires the field to have a minimum number of elements. The schema will have a
minItems
validation keywordmax_items – only applies to lists, requires the field to have a maximum number of elements. The schema will have a
maxItems
validation keywordunique_items – only applies to lists, requires the field not to have duplicated elements. The schema will have a
uniqueItems
validation keywordmin_length – only applies to strings, requires the field to have a minimum length. The schema will have a
minLength
validation keywordmax_length – only applies to strings, requires the field to have a maximum length. The schema will have a
maxLength
validation keywordallow_mutation – a boolean which defaults to True. When False, the field raises a TypeError if the field is assigned on an instance. The BaseModel Config must set validate_assignment to True
regex – only applies to strings, requires the field match against a regular expression pattern string. The schema will have a
pattern
validation keyworddiscriminator – only useful with a (discriminated a.k.a. tagged) Union of sub models with a common field. The discriminator is the name of this common field to shorten validation and improve generated schema
repr – show this field in the representation
**extra – any additional keyword arguments will be added as is to the schema
- blacksmith.QueryStringField(default: Any = PydanticUndefined, *, default_factory: Optional[Callable[[], Any]] = None, alias: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny, Any]] = None, const: Optional[bool] = None, gt: Optional[float] = None, ge: Optional[float] = None, lt: Optional[float] = None, le: Optional[float] = None, multiple_of: Optional[float] = None, allow_inf_nan: Optional[bool] = None, max_digits: Optional[int] = None, decimal_places: Optional[int] = None, min_items: Optional[int] = None, max_items: Optional[int] = None, unique_items: Optional[bool] = None, min_length: Optional[int] = None, max_length: Optional[int] = None, allow_mutation: bool = True, regex: Optional[str] = None, discriminator: Optional[str] = None, repr: bool = True, **extra: Any) Any #
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 tostr
.- Parameters
default – since this is replacing the field’s default, its first argument is used to set the default, use ellipsis (
...
) to indicate the field is requireddefault_factory – callable that will be called when a default value is needed for this field If both default and default_factory are set, an error is raised.
alias – the public name of the field
title – can be any string, used in the schema
description – can be any string, used in the schema
exclude – exclude this field while dumping. Takes same values as the
include
andexclude
arguments on the.dict
method.include – include this field while dumping. Takes same values as the
include
andexclude
arguments on the.dict
method.const – this field is required and must take it’s default value
gt – only applies to numbers, requires the field to be “greater than”. The schema will have an
exclusiveMinimum
validation keywordge – only applies to numbers, requires the field to be “greater than or equal to”. The schema will have a
minimum
validation keywordlt – only applies to numbers, requires the field to be “less than”. The schema will have an
exclusiveMaximum
validation keywordle – only applies to numbers, requires the field to be “less than or equal to”. The schema will have a
maximum
validation keywordmultiple_of – only applies to numbers, requires the field to be “a multiple of”. The schema will have a
multipleOf
validation keywordallow_inf_nan – only applies to numbers, allows the field to be NaN or infinity (+inf or -inf), which is a valid Python float. Default True, set to False for compatibility with JSON.
max_digits – only applies to Decimals, requires the field to have a maximum number of digits within the decimal. It does not include a zero before the decimal point or trailing decimal zeroes.
decimal_places – only applies to Decimals, requires the field to have at most a number of decimal places allowed. It does not include trailing decimal zeroes.
min_items – only applies to lists, requires the field to have a minimum number of elements. The schema will have a
minItems
validation keywordmax_items – only applies to lists, requires the field to have a maximum number of elements. The schema will have a
maxItems
validation keywordunique_items – only applies to lists, requires the field not to have duplicated elements. The schema will have a
uniqueItems
validation keywordmin_length – only applies to strings, requires the field to have a minimum length. The schema will have a
minLength
validation keywordmax_length – only applies to strings, requires the field to have a maximum length. The schema will have a
maxLength
validation keywordallow_mutation – a boolean which defaults to True. When False, the field raises a TypeError if the field is assigned on an instance. The BaseModel Config must set validate_assignment to True
regex – only applies to strings, requires the field match against a regular expression pattern string. The schema will have a
pattern
validation keyworddiscriminator – only useful with a (discriminated a.k.a. tagged) Union of sub models with a common field. The discriminator is the name of this common field to shorten validation and improve generated schema
repr – show this field in the representation
**extra – any additional keyword arguments will be added as is to the schema
- 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: typing_extensions.Literal[HEAD, GET, POST, PUT, PATCH, DELETE, OPTIONS], path: str, name: str, client_name: str, error_parser: blacksmith.domain.error.AbstractErrorParser[blacksmith.domain.error.TError_co])#
Wrap a HTTP response and deserialize it.
user: ResponseBox[User, HTTPError] = ( await api.user.get({"username": username}) ) if user.is_ok(): print(user.unwrap().username) else: print(f"API Call failed: {user.unwrap_err()}")
- property json: Optional[Dict[str, Any]]#
Return the raw json response.
It return the raw response body without noticing if its a normal or an error response.
- property response: blacksmith.domain.model.params.TResponse#
Parse the response using the schema.
Deprecated since version 2.0: Use
ResponseBox.unwrap()
- Raises
blacksmith.HTTPError – if the response contains an error.
NoResponseSchemaException – if the response_schema has not been set in the contract.
- as_result() Union[result.result.Ok[blacksmith.domain.model.params.TResponse], result.result.Err[blacksmith.domain.error.TError_co]] #
Return the result as a
result.Result
.The
blacksmith.ResponseBox
mimic theresult.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 anyblacksmith.NoResponseSchemaException
- is_ok() bool #
Return True if the response was an http success.
- is_err() bool #
Return True if the response was an http error.
- unwrap() blacksmith.domain.model.params.TResponse #
Return the parsed response.
- Raises
NoResponseSchemaException – if there are no response schema set.
- unwrap_err() blacksmith.domain.error.TError_co #
Return the response error.
- unwrap_or(default: blacksmith.domain.model.params.TResponse) blacksmith.domain.model.params.TResponse #
Return the response or the default value in case of error.
- Raises
NoResponseSchemaException – if there are no response schema set.
- unwrap_or_else(op: Callable[[blacksmith.domain.error.TError_co], blacksmith.domain.model.params.TResponse]) blacksmith.domain.model.params.TResponse #
Return the response or the callable return in case of error.
- Raises
NoResponseSchemaException – if there are no response schema set.
- expect(message: str) blacksmith.domain.model.params.TResponse #
Return the response or raise an UnwrapError exception with the given message.
- Raises
NoResponseSchemaException – if there are no response schema set.
- expect_err(message: str) blacksmith.domain.error.TError_co #
Return the error or raise an UnwrapError exception with the given message.
- map(op: Callable[[blacksmith.domain.model.params.TResponse], result.result.U]) Union[result.result.Ok[result.result.U], result.result.Err[blacksmith.domain.error.TError_co]] #
Apply op on response in case of success, and return the new result.
- Raises
NoResponseSchemaException – if there are no response schema set.
- map_or(default: result.result.U, op: Callable[[blacksmith.domain.model.params.TResponse], result.result.U]) result.result.U #
Apply and return op on response in case of success, default in case of error.
- Raises
NoResponseSchemaException – if there are no response schema set.
- map_or_else(default_op: Callable[[], result.result.U], op: Callable[[blacksmith.domain.model.params.TResponse], result.result.U]) result.result.U #
Return the result of default_op in case of error otherwise the result of op.
- Raises
NoResponseSchemaException – if there are no response schema set.
- map_err(op: Callable[[blacksmith.domain.exceptions.HTTPError], result.result.F]) Union[result.result.Ok[blacksmith.domain.model.params.TResponse], result.result.Err[result.result.F]] #
Apply op on error in case of error, and return the new result.
- Raises
NoResponseSchemaException – if there are no response schema set.
- and_then(op: Callable[[blacksmith.domain.model.params.TResponse], Union[result.result.Ok[result.result.U], result.result.Err[blacksmith.domain.exceptions.HTTPError]]]) Union[result.result.Ok[result.result.U], result.result.Err[blacksmith.domain.exceptions.HTTPError]] #
Apply the op function on the response and return it if success
- Raises
NoResponseSchemaException – if there are no response schema set.
- or_else(op: Callable[[blacksmith.domain.exceptions.HTTPError], Union[result.result.Ok[blacksmith.domain.model.params.TResponse], result.result.Err[result.result.F]]]) Union[result.result.Ok[blacksmith.domain.model.params.TResponse], result.result.Err[result.result.F]] #
Apply the op function on the error and return it if error
- Raises
NoResponseSchemaException – if there are no response schema set.
- class blacksmith.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.
- 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.
- 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.
- 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
- registry: blacksmith.domain.registry.Registry#
- 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
- registry: blacksmith.domain.registry.Registry#
- 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]#
- 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]#
- 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#
- 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 theblacksmith.AsyncClientFactory
( orblacksmith.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#
- 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 theblacksmith.AsyncClientFactory
( orblacksmith.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.
- headers: Dict[str, str]#
- 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.
- headers: Dict[str, str]#
- class blacksmith.AsyncHTTPBearerMiddleware(bearer_token: str)#
Authentication Mechanism based on the header Authorization with the Bearer scheme.
- Parameters
value – value of the bearer token.
- headers: Dict[str, str]#
- class blacksmith.SyncHTTPBearerMiddleware(bearer_token: str)#
Authentication Mechanism based on the header Authorization with the Bearer scheme.
- Parameters
value – value of the bearer token.
- headers: Dict[str, str]#
- class blacksmith.AsyncCircuitBreakerMiddleware(threshold: int = 5, ttl: float = 30, listeners: Optional[Iterable[Callable[[str, typing_extensions.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, typing_extensions.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.
- 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.
- 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: typing_extensions.Literal[uncachable_request, uncachable_response, cached], method: typing_extensions.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: typing_extensions.Literal[uncachable_request, uncachable_response, cached], method: typing_extensions.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: typing_extensions.Literal[HEAD, GET, POST, PUT, PATCH, DELETE, OPTIONS], url_pattern: str, path: Dict[str, Union[str, int, float, bool]] = <factory>, querystring: Dict[str, Union[str, int, float, bool, List[Union[str, int, float, bool]]]] = <factory>, headers: Dict[str, str] = <factory>, body: str = '')#
Internal representation of an http request.
Note that the HTTP method is not present, because the method is the funcion called.
The HTTP Request is filled out using the
blacksmith.domain.model.params.Request
schema.- method: typing_extensions.Literal[HEAD, GET, POST, PUT, PATCH, DELETE, OPTIONS]#
- url_pattern: str#
- path: Dict[str, Union[str, int, float, bool]]#
- querystring: Dict[str, Union[str, int, float, bool, List[Union[str, int, float, bool]]]]#
- headers: Dict[str, str]#
- body: str = ''#
- property url: str#
- class blacksmith.HTTPResponse(status_code: int, headers: Mapping[str, str], json: Optional[Any])#
Internal representation of an http response.
- status_code: int#
HTTP Status code.
- headers: Mapping[str, str]#
Header of the response.
- json: Optional[Any]#
Json Body of the response.
- property links: Dict[Optional[str], Dict[str, str]]#