Consul#
The discovery based on Consul.
This driver implement a client side service discovery.
- exception blacksmith.sd._async.adapters.consul.ConsulApiError(exc: blacksmith.domain.exceptions.HTTPError)#
Raised when consul API is not responding what is expected.
- class blacksmith.sd._async.adapters.consul.ServiceRequest(*, name: str)#
Request parameter of the Consul API to retrieve a host for a service.
- name: str#
Name of the service to search for an endpoint.
- model_config: ClassVar[ConfigDict] = {}#
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- model_fields: ClassVar[dict[str, FieldInfo]] = {'name': FieldInfo(annotation=str, required=True, json_schema_extra={'location': 'path'})}#
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.sd._async.adapters.consul.Service(*, Address: str, ServiceAddress: Optional[str] = None, ServicePort: int)#
Consul Service response.
- node_address: str#
IP address of the Consul node on which the service is registered.
- service_address: Optional[str]#
IP address of the service host. if empty, node address is used.
- port: int#
TCP Port of an instance that host the service.
- property address: str#
- model_config: ClassVar[ConfigDict] = {}#
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- model_fields: ClassVar[dict[str, FieldInfo]] = {'node_address': FieldInfo(annotation=str, required=True, alias='Address', alias_priority=2), 'port': FieldInfo(annotation=int, required=True, alias='ServicePort', alias_priority=2), 'service_address': FieldInfo(annotation=Union[str, NoneType], required=False, alias='ServiceAddress', alias_priority=2)}#
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.sd._async.adapters.consul.blacksmith_cli(endpoint: str, consul_token: str) blacksmith.service._async.client.AsyncClientFactory[blacksmith.domain.exceptions.HTTPError] #
- class blacksmith.sd._async.adapters.consul.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.