(
self,
*,
persistence_enabled: bool | Literal['explicit_only'] = False,
persist_state_kvs_name: str | None = None,
persist_state_key: str | None = None,
persist_state_kvs_factory: Callable[[], Coroutine[None, None, KeyValueStore]] | None = None,
log_message: str = 'Statistics',
periodic_message_logger: Logger | None = None,
log_interval: timedelta = timedelta(minutes=1),
state_model: type[TStatisticsState],
statistics_log_format: Literal['table', 'inline'] = 'table',
save_error_snapshots: bool = False,
)
| 69 | __next_id = 0 |
| 70 | |
| 71 | def __init__( |
| 72 | self, |
| 73 | *, |
| 74 | persistence_enabled: bool | Literal['explicit_only'] = False, |
| 75 | persist_state_kvs_name: str | None = None, |
| 76 | persist_state_key: str | None = None, |
| 77 | persist_state_kvs_factory: Callable[[], Coroutine[None, None, KeyValueStore]] | None = None, |
| 78 | log_message: str = 'Statistics', |
| 79 | periodic_message_logger: Logger | None = None, |
| 80 | log_interval: timedelta = timedelta(minutes=1), |
| 81 | state_model: type[TStatisticsState], |
| 82 | statistics_log_format: Literal['table', 'inline'] = 'table', |
| 83 | save_error_snapshots: bool = False, |
| 84 | ) -> None: |
| 85 | self._id = Statistics.__next_id |
| 86 | Statistics.__next_id += 1 |
| 87 | |
| 88 | self.error_tracker = ErrorTracker( |
| 89 | save_error_snapshots=save_error_snapshots, |
| 90 | snapshot_kvs_name=persist_state_kvs_name, |
| 91 | ) |
| 92 | self.error_tracker_retry = ErrorTracker(save_error_snapshots=False) |
| 93 | |
| 94 | self._requests_in_progress = dict[str, RequestProcessingRecord]() |
| 95 | |
| 96 | self._state = RecoverableState( |
| 97 | default_state=state_model(stats_id=self._id), |
| 98 | persist_state_key=persist_state_key or f'__CRAWLER_STATISTICS_{self._id}', |
| 99 | persistence_enabled=persistence_enabled, |
| 100 | persist_state_kvs_name=persist_state_kvs_name, |
| 101 | persist_state_kvs_factory=persist_state_kvs_factory, |
| 102 | logger=logger, |
| 103 | ) |
| 104 | |
| 105 | self._log_message = log_message |
| 106 | self._statistics_log_format = statistics_log_format |
| 107 | self._periodic_message_logger = periodic_message_logger or logger |
| 108 | self._periodic_logger = RecurringTask(self._log, log_interval) |
| 109 | |
| 110 | # Flag to indicate the context state. |
| 111 | self._active = False |
| 112 | |
| 113 | def replace_state_model(self, state_model: type[TNewStatisticsState]) -> Statistics[TNewStatisticsState]: |
| 114 | """Create near copy of the `Statistics` with replaced `state_model`.""" |
nothing calls this directly
no test coverage detected