Consul

The discovery based on Consul.

This driver implement a client side service discovery.

exception blacksmith.sd._async.adapters.consul.ConsulApiError(exc: 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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

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

This replaces Model.__fields__ from Pydantic V1.

class blacksmith.sd._async.adapters.consul.Service(*, Address: str, ServiceAddress: str | None = None, ServicePort: int)

Consul Service response.

node_address: str

IP address of the Consul node on which the service is registered.

service_address: str | None

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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

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, default=None, alias='ServiceAddress', alias_priority=2)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

blacksmith.sd._async.adapters.consul.blacksmith_cli(endpoint: str, consul_token: str) → AsyncClientFactory[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: ~typing.Callable[[str, str], ~blacksmith.service._async.client.AsyncClientFactory[~typing.Any]] = <function blacksmith_cli>)

A discovery instance based on a Consul server.

Parameters:

  • addr – endpoint of the consul v1 http api.

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

  • consul_token – If set, the consul token is sent on http api call.

addr: str
consul_token: str
service_name_fmt: str
service_url_fmt: str
unversioned_service_name_fmt: str
unversioned_service_url_fmt: str
format_service_name(service: str, version: str | None) → str

Build the service name to send to consul.

format_endoint(version: str | None, address: str, port: int) → str

Build the rest api endpoint from consul response.

async resolve(service: str, version: str | None) → 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: str | None) → str

Get the endpoint from the consul registry

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