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_config: ClassVar[ConfigDict] = {}

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

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_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

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][source]
model_config: ClassVar[ConfigDict] = {}

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

class ProxySettings[source]

Bases: ApplicationSettingsMixin, 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_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

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_config: ClassVar[ConfigDict] = {}

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

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_config: ClassVar[ConfigDict] = {}

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

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_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

pools: List[str]
url: HttpUrl
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_config: ClassVar[ConfigDict] = {}

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

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_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

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_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

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

probe: RemoteProbeSettings | None

Submodules