configs

configs

Synchronizer service configuration models.

See Also

Synchronizer: The service class that consumes these configurations. BaseServiceConfig: Base class providing interval and log_level fields.

Classes

TimeoutsConfig

Bases: BaseModel

Sync timeout limits: idle progress check and phase-level cap.

idle controls the progress-based idle timeout per relay: if stream_events yields no events for this many seconds the relay is abandoned and the semaphore slot freed. The timer resets on every yielded event, so a relay that produces events slowly but steadily is never killed.

max_duration caps the entire sync phase: once exceeded, remaining relays are skipped.

See Also

SynchronizerConfig: Parent config that embeds this model.

ProcessingConfig

Bases: BaseModel

Sync processing parameters following NIP-01 REQ semantics.

Attributes:

• filters (list[Filter]) –

  NIP-01 filter dicts, converted to nostr_sdk.Filter at load time for fail-fast validation.

• since (int) –

  Default start timestamp for relays without a cursor.

• until (int | None) –

  Upper bound; None (default) means now().

• limit (int) –

  Max events per relay request (REQ limit).

• end_lag (int) –

  Seconds subtracted from until to compute the actual sync end time: (until or now()) - end_lag.

• allow_insecure (bool) –

  Fall back to insecure transport on SSL failure.

See Also

SynchronizerConfig: Parent config that embeds this model.

Functions

parse_filters classmethod

parse_filters(v: Any) -> list[Filter]

Convert raw NIP-01 filter dicts to nostr_sdk.Filter objects.

Source code in src/bigbrotr/services/synchronizer/configs.py

@field_validator("filters", mode="before")
@classmethod
def parse_filters(cls, v: Any) -> list[Filter]:
    """Convert raw NIP-01 filter dicts to ``nostr_sdk.Filter`` objects."""
    if not isinstance(v, list):
        raise TypeError(f"filters: expected list, got {type(v).__name__}")
    return [_parse_filter(raw, i) for i, raw in enumerate(v)]

get_end_time

get_end_time() -> int

Compute the sync end timestamp: (until or now()) - end_lag.

Source code in src/bigbrotr/services/synchronizer/configs.py

def get_end_time(self) -> int:
    """Compute the sync end timestamp: ``(until or now()) - end_lag``."""
    base = self.until if self.until is not None else int(time.time())
    return base - self.end_lag

SynchronizerConfig

Bases: BaseServiceConfig

Synchronizer service configuration.

See Also

Synchronizer: The service class that consumes this configuration. BaseServiceConfig: Base class providing interval, max_consecutive_failures, and metrics fields. NetworksConfig: Per-network timeout and proxy settings. KeysConfig: Nostr key management for NIP-42 authentication during event fetching.