harp_apps.proxy.settings.liveness.leaky_bucket

class LeakyBucketLiveness

Bases: BaseLiveness[LeakyBucketLivenessSettings]

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(subject, reason=None)

Parameters:

• subject (LivenessSubject)

• reason (str | None)

Return type:

bool

classmethod get_state_of(subject)

Parameters:

subject (LivenessSubject)

Return type:

LeakyBucketLivenessSubjectState

success(subject)

Parameters:

subject (LivenessSubject)

Return type:

bool

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=LeakyBucketLivenessSettings, 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 LeakyBucketLivenessSettings

Bases: BaseLivenessSettings

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.

build_impl()

capacity: float

The maximum capacity of the bucket.

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]] = {'capacity': FieldInfo(annotation=float, required=False, default=60.0), 'rate': FieldInfo(annotation=float, required=False, default=1.0), 'threshold': FieldInfo(annotation=float, required=False, default=10.0), 'type': FieldInfo(annotation=Literal['leaky'], required=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.

rate: float

The rate at which the bucket leaks (x per second).

threshold: float

The threshold over which the remote is considered down.

type: Literal['leaky']

class LeakyBucketLivenessSubjectState

Bases: BaseModel

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.

leak(rate)

Parameters:

rate (float)

current: float

last_checked: float | None

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]] = {'current': FieldInfo(annotation=float, required=False, default=0.0), 'last_checked': FieldInfo(annotation=Union[float, 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.