webchanges.handler module

Handles the running of jobs and, afterward, of the reports.

class webchanges.handler.Snapshot(data, timestamp, tries, etag, mime_type, error_data)

Bases: NamedTuple

Type for Snapshot named tuple.

0: data: str | bytes
1: timestamp: float
2: tries: int
3: etag: str
4: mime_type: mime_type
5: error: ErrorData

Create new instance of Snapshot(data, timestamp, tries, etag, mime_type, error_data)

Parameters:

data (str | bytes)
timestamp (float)
tries (int)
etag (str)
mime_type (str)
error_data (ErrorData)

data: str | bytes: Alias for field number 0

timestamp: float: Alias for field number 1

tries: int: Alias for field number 2

etag: str: Alias for field number 3

mime_type: str: Alias for field number 4

error_data: ErrorData: Alias for field number 5

count(value, /): Return number of occurrences of value.

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

class webchanges.handler.ErrorData

Bases: TypedDict

clear(): Remove all items from the dict.

copy(): Return a shallow copy of the dict.

classmethod fromkeys(iterable, value=None, /): Create a new dictionary with keys from iterable and values set to value.

get(key, default=None, /): Return the value for key if key is in the dictionary, else default.

items(): Return a set-like object providing a view on the dict’s items.

keys(): Return a set-like object providing a view on the dict’s keys.

pop(k[, d]) → v, remove specified key and return the corresponding value.: If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

setdefault(key, default=None, /)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None. Update D from mapping/iterable E and F.: If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values(): Return an object providing a view on the dict’s values.

type: str

message: str

class webchanges.handler.JobState(snapshots_db, job)

Bases: ContextManager

The JobState class, which contains run information about a job.

Initializes the class

Parameters:

snapshots_db (SsdbStorage) – The SsdbStorage object with the snapshot database methods.
job (JobBase) – A JobBase object with the job information.

error_ignored: bool

exception: Exception | None = None

new_data: str | bytes = ''

new_error_data: ErrorData = {}

new_etag: str = ''

new_mime_type: str = ''

new_timestamp: float

old_snapshot = ('', 1605147837.511478, 0, '', 'text/plain', {})

old_data: str | bytes = ''

old_error_data: ErrorData = {}

old_etag: str = ''

old_mime_type: str = 'text/plain'

old_timestamp: float = 1605147837.511478

traceback: str

tries: int = 0

verb: Verb

job: JobBase

generated_diff: dict[ReportKind, str]

unfiltered_diff: dict[ReportKind, str]

history_dic_snapshots: dict[str | bytes, Snapshot]

static debugging_session()

Checks if the code is currently running within an external debugger (e.g. IDE) and NOT in a testing environment.

Returns:: True if an external debugger is attached and it’s not a pytest session, False otherwise.
Return type:: bool

added_data()

Returns a dict with the data added in the processing of the job.

Return type:: dict[str, bool | str | Exception | float | None]

load()

Loads form the database the last snapshot(s) for the job.

Return type:: None

save()

Saves new data retrieved by the job into the snapshot database.

Return type:: None

delete_latest(temporary=True)

Removes the last instance in the snapshot database.

Parameters:: temporary (bool)
Return type:: None

process(headless=True)

Processes the job: loads it (i.e. runs it) and handles Exceptions (errors).

Returns:: a JobState object containing information of the job run.
Parameters:: headless (bool)
Return type:: JobState

get_diff(report_kind='plain', differ=None, differ_defaults=None, tz=None)

Generates the job’s diff and applies diff_filters to it (if any). Memoized.

Parameters:

report_kind (ReportKind) – the kind of report that needs the differ.
differ (dict[str, Any] | None) – the name of the differ to override self.job.differ.
tz (ZoneInfo | None) – The IANA tz_info name of the timezone to use for diff in the job’s report (e.g. ‘Etc/UTC’).
report_kind
differ
differ_defaults (_ConfigDifferDefaults | None)
tz

Returns:

The job’s diff.

Return type:

str

is_markdown()

Returns whether the new data is in markdown.

Return type:: bool

class webchanges.handler.Report(urlwatch)

Bases: object

The base class for reporting.

Parameters:: urlwatch (Urlwatch) – The Urlwatch object with the program configuration information.

job_states: list[JobState] = []

new_release_future: Future[str | bool] | None = None

start: float = 36613.361383371

config: _Config

new(job_state)

Sets the verb of the job in job_state to ‘new’. Called by run_jobs() and tests.

Parameters:: job_state (JobState) – The JobState object with the information of the job run.
Return type:: None

changed(job_state)

Sets the verb of the job in job_state to ‘changed’. Called by run_jobs() and tests.

Parameters:: job_state (JobState) – The JobState object with the information of the job run.
Return type:: None

changed_no_report(job_state)

Sets the verb of the job in job_state to ‘changed,no_report’. Called by run_jobs() and tests.

Parameters:: job_state (JobState) – The JobState object with the information of the job run.
Return type:: None

unchanged(job_state)

Sets the verb of the job in job_state to ‘unchanged’. Called by run_jobs() and tests.

Parameters:: job_state (JobState) – The JobState object with the information of the job run.
Return type:: None

unchanged_from_error(job_state)

Sets the verb of the job in job_state to ‘unchanged’. Called by run_jobs() and tests.

Parameters:: job_state (JobState) – The JobState object with the information of the job run.
Return type:: None

error(job_state)

Sets the verb of the job in job_state to ‘error’. Called by run_jobs() and tests.

Parameters:: job_state (JobState) – The JobState object with the information of the job run.
Return type:: None

error_same_error(job_state)

Sets the verb of the job in job_state to ‘error’. Called by run_jobs() and tests.

Parameters:: job_state (JobState) – The JobState object with the information of the job run.
Return type:: None

custom(job_state, label)

Sets the verb of the job in job_state to a custom label. Called by UrlwatchCommand.check_test_reporter().

Parameters:

job_state (JobState) – The JobState object with the information of the job run.
label (Literal['new', 'changed', 'changed,no_report', 'unchanged', 'unchanged,error_ended', 'error', 'error,repeated']) – The label to set the information of the job run to.

Return type:

None

get_filtered_job_states(job_states)

Returns JobStates that have reportable changes per config[‘display’]. Called from ReporterBase.

Parameters:: job_states (list[JobState]) – The list of JobState objects with the information of the job runs.
Returns:: An iterable of JobState objects that have reportable changes per config[‘display’].
Return type:: Iterator[JobState]

finish(jobs_file)

Finish job run: determine its duration and generate reports by submitting job_states to ReporterBase submit_all().

Parameters:: jobs_file (list[Path]) – The path to the file containing the list of jobs (optional, used in footers).
Return type:: None

finish_one(name, jobs_file=None, check_enabled=True)

Finish job run of one: determine its duration and generate reports by submitting job_states to ReporterBase submit_one(). Used in testing.

Parameters:

name (str) – The name of the reporter to run.
jobs_file (list[Path] | None) – The path to the file containing the list of jobs (optional, used in footers).
check_enabled (bool | None) – If True (default), run reports only if they are enabled in the configuration.

Return type:

None