harp_apps.proxy.settings

Inheritance diagram of harp_apps.proxy.settings

class Endpoint[source]

Bases: Stateful[EndpointSettings]

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

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]] = {'remote': FieldInfo(annotation=Remote, required=False, default=None), 'settings': FieldInfo(annotation=EndpointSettings, required=True, repr=False)}

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.

remote: Remote
class EndpointSettings[source]

Bases: BaseEndpointSettings

Configuration parser for proxy.endpoints[] settings.

name: my-endpoint
port: 8080
description: My endpoint
remote:
  # see HttpRemote
  ...
controller : optional controller Service definition, default to HttpProxyController

A shorthand syntax is also available for cases where you only need to proxy to a single URL and do not require fine-tuning the endpoint settings:

name: my-endpoint
port: 8080
description: My endpoint
url: http://my-endpoint:8080

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

controller: Service | str | None

custom controller

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

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

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'controller': FieldInfo(annotation=Union[Service, str, NoneType], required=False, default=Service(base=None, type='harp_apps.proxy.controllers.HttpProxyController', constructor=None, arguments={'dispatcher': LazyService(type='IAsyncEventDispatcher')})), 'description': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'name': FieldInfo(annotation=str, required=True), 'port': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'remote': FieldInfo(annotation=Union[RemoteSettings, NoneType], required=False, default=None, repr=False)}

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.

remote: RemoteSettings | None

remote definition, with url pools, probes, etc.

class Proxy[source]

Bases: Stateful[ProxySettings]

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

property endpoints: list[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]] = {'settings': FieldInfo(annotation=ProxySettings, required=True, repr=False)}

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 ProxySettings[source]

Bases: BaseProxySettings

Configuration parser for proxy settings.

endpoints:
  # see ProxyEndpoint
  - ...

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

endpoints: list[EndpointSettings]
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

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

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'endpoints': FieldInfo(annotation=list[EndpointSettings], required=False, default=[])}

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 Remote[source]

Bases: Stateful[RemoteSettings]

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async check()[source]

Uses the probe (luke), to check the health of each urls. It is also done on fallback and inactive urls, to ensure that they are ready in case we need them.

async check_forever()[source]
get_url()[source]

Get next candidate url from the current pool, then rotate.

Return type:

str

model_post_init(context, /)

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

Parameters:

  • self (BaseModel)

  • context (Any)

Return type:

None

notify_url_status(url, status)[source]

Take into account a http status code to update the status of the url. Behaviour differs depending on the instance configuration (some considers 5xx as down, some ignores non-network errors).

refresh()[source]

Recompute the current pool of endpoints.

set_checking(url)[source]
set_down(url)[source]
set_up(url)[source]
DEFAULT_LIVENESS: ClassVar[BaseLiveness] = IgnoreLiveness()

Default liveness object to use when the remote is set to inherit the liveness object.

property current_pool: List[str]
current_pool_name: str

Name of the currently used pool. This does not mean that all urls come from this pool, as the fallback pool may be active although some urls from default pool are still available.

property endpoints: List[RemoteEndpoint]
liveness: IgnoreLiveness | InheritLiveness | LeakyBucketLiveness | NaiveLiveness

Liveness

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {'current_pool': ComputedFieldInfo(wrapped_property=<property object>, return_type=typing.List[str], alias=None, alias_priority=None, title=None, field_title_generator=None, description=None, deprecated=None, examples=None, json_schema_extra=None, repr=True), 'endpoints': ComputedFieldInfo(wrapped_property=<property object>, return_type=typing.List[harp_apps.proxy.settings.remote.endpoint.RemoteEndpoint], alias=None, alias_priority=None, title=None, field_title_generator=None, description=None, deprecated=None, examples=None, json_schema_extra=None, repr=True)}

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]] = {'current_pool_name': FieldInfo(annotation=str, required=False, default='default'), 'liveness': FieldInfo(annotation=Union[IgnoreLiveness, InheritLiveness, LeakyBucketLiveness, NaiveLiveness], required=False, default=None, exclude=True), 'probe': FieldInfo(annotation=Union[RemoteProbe, NoneType], required=False, default=None), 'settings': FieldInfo(annotation=RemoteSettings, required=True, repr=False)}

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.

probe: RemoteProbe | None

Probe reference

class RemoteEndpoint[source]

Bases: Stateful[RemoteEndpointSettings]

Stateful version of a remote endpoint definition.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

failure(reason=None)[source]

Returns a boolean indicating if a state change happened.

Parameters:

reason (str)

success()[source]

Returns a boolean indicating if a state change happened.

Return type:

bool

failure_reasons: set | None
liveness: IgnoreLiveness | InheritLiveness | LeakyBucketLiveness | NaiveLiveness
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]] = {'failure_reasons': FieldInfo(annotation=Union[set, NoneType], required=False, default=None), 'liveness': FieldInfo(annotation=Union[IgnoreLiveness, InheritLiveness, LeakyBucketLiveness, NaiveLiveness], required=False, default=None, exclude=True), 'settings': FieldInfo(annotation=RemoteEndpointSettings, required=True, repr=False), 'status': FieldInfo(annotation=int, required=False, default=0)}

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.

status: int
class RemoteEndpointSettings[source]

Bases: Configurable

A HttpEndpoint is an instrumented target URL that a proxy will use to route requests. It is used as the configuration parser for proxy.endpoints[].remote.endpoints[] settings.

url: "http://my-endpoint:8080"
pools: [fallback]  # omit for default pool
failure_threshold: 3
success_threshold: 1

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

classmethod serialize_in_order(value)[source]
Parameters:

value (List[str])

classmethod validate_pools(pools)[source]
Parameters:

pools (List[str])

Return type:

List[str]

liveness: Annotated[IgnoreLivenessSettings | InheritLivenessSettings | LeakyBucketLivenessSettings | NaiveLivenessSettings, Discriminator(discriminator=type, custom_error_type=None, custom_error_message=None, custom_error_context=None)]
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

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

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'liveness': FieldInfo(annotation=Union[IgnoreLivenessSettings, InheritLivenessSettings, LeakyBucketLivenessSettings, NaiveLivenessSettings], required=False, default_factory=InheritLivenessSettings, metadata=[Discriminator(discriminator='type', custom_error_type=None, custom_error_message=None, custom_error_context=None)]), 'pools': FieldInfo(annotation=List[str], required=False, default=['default']), 'url': FieldInfo(annotation=Url, required=True, metadata=[UrlConstraints(max_length=2083, allowed_schemes=['http', 'https'], host_required=None, default_host=None, default_port=None, default_path=None)])}

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.

pools: List[str]
url: Annotated[Url, UrlConstraints(max_length=2083, allowed_schemes=['http', 'https'], host_required=None, default_host=None, default_port=None, default_path=None)]
class RemoteProbe[source]

Bases: Stateful[RemoteProbeSettings]

Stateful version of a probe definition.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

async_client()[source]
async check(client, endpoint)[source]
Parameters:
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]] = {'settings': FieldInfo(annotation=RemoteProbeSettings, required=True, repr=False)}

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 RemoteProbeSettings[source]

Bases: Configurable

A HttpProbe is a health check that can be used to check the health of a remote’s endpoints. It is used as the configuration parser for proxy.endpoints[].remote.probe settings.

type: http
method: GET
path: /health
headers:
  x-purpose: "health probe"
timeout: 5.0

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

headers: dict
interval: float
method: str
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

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

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'headers': FieldInfo(annotation=dict, required=False, default_factory=dict), 'interval': FieldInfo(annotation=float, required=False, default=10.0), 'method': FieldInfo(annotation=str, required=False, default='GET'), 'path': FieldInfo(annotation=str, required=False, default='/'), 'timeout': FieldInfo(annotation=float, required=False, default=10.0), 'verify': FieldInfo(annotation=bool, required=False, default=True)}

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.

path: str
timeout: float
verify: bool
class RemoteSettings[source]

Bases: BaseRemoteSettings

A HttpRemote is a collection of endpoints that a proxy will use to route requests. It is used as the configuration parser for proxy.endpoints[].remote settings.

min_pool_size: 1
endpoints:
  # see HttpEndpoint
  - ...
probe:
  # see HttpProbe
  ...

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

endpoints: list[RemoteEndpointSettings]
liveness: Annotated[IgnoreLivenessSettings | InheritLivenessSettings | LeakyBucketLivenessSettings | NaiveLivenessSettings, Discriminator(discriminator=type, custom_error_type=None, custom_error_message=None, custom_error_context=None)]
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

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

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'break_on': FieldInfo(annotation=list[str], required=False, default=['network_error', 'unhandled_exception']), 'check_after': FieldInfo(annotation=float, required=False, default=10.0), 'endpoints': FieldInfo(annotation=list[RemoteEndpointSettings], required=False, default=None), 'liveness': FieldInfo(annotation=Union[IgnoreLivenessSettings, InheritLivenessSettings, LeakyBucketLivenessSettings, NaiveLivenessSettings], required=False, default=InheritLivenessSettings(type='inherit'), metadata=[Discriminator(discriminator='type', custom_error_type=None, custom_error_message=None, custom_error_context=None)]), 'min_pool_size': FieldInfo(annotation=int, required=False, default=1), 'probe': FieldInfo(annotation=Union[RemoteProbeSettings, NoneType], required=False, default=None)}

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.

probe: RemoteProbeSettings | None

Submodules