21856489627

Committed 10 Feb 2026 07:57AM UTC coverage: 73.228% (-0.09%) from 73.318%

Build # 21856489627

Build Type

push

github

Committed by

mborsetti

Commit Message

Version 3.34.0rc0

Coverage Stats

1424 of 2298 branches covered (61.97%)

Branch coverage included in aggregate %.

4766 of 6155 relevant lines covered (77.43%)

11.07 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.79

/webchanges/main.py

"""The main class.

For the entry point, see main() in the cli module.
"""

# The code below is subject to the license contained in the LICENSE.md file, which is part of the source code.

from __future__ import annotations

import logging
from typing import TYPE_CHECKING

from webchanges import __project_name__
from webchanges.handler import Report
from webchanges.util import get_new_version_number
from webchanges.worker import run_jobs

if TYPE_CHECKING:
    from webchanges.config import CommandConfig
    from webchanges.jobs import JobBase
    from webchanges.storage import SsdbStorage, YamlConfigStorage, YamlJobsStorage


logger = logging.getLogger(__name__)


class Urlwatch:
    """The main class."""

    config_storage: YamlConfigStorage
    ssdb_storage: SsdbStorage
    jobs_storage: YamlJobsStorage
    jobs: list[JobBase]
    report: Report

    def __init__(
        self,
        urlwatch_config: CommandConfig,
        config_storage: YamlConfigStorage,
        ssdb_storage: SsdbStorage,
        jobs_storage: YamlJobsStorage,
    ) -> None:
        """:param urlwatch_config: The CommandConfig object containing the program run information.
        :param config_storage: The YamlConfigStorage object containing the configuration information.
        :param ssdb_storage: The CacheStorage object containing snapshot database information
        :param jobs_storage: The YamlJobsStorage object containing information about the jobs.
        """
        self.urlwatch_config = urlwatch_config

        self.config_storage = config_storage
        self.ssdb_storage = ssdb_storage
        self.jobs_storage = jobs_storage

        self.report = Report(self)
        self.jobs: list[JobBase] = []

        self._latest_release: str | bool | None = None

        self.check_directories()

        if not self.urlwatch_config.edit:
            self.load_jobs()

    def check_directories(self) -> None:
        """Check whether the configuration and jobs files directories exist. Create them and write default configuration
        files if one not found.
        """
        if (
            not (self.urlwatch_config.config_file and self.urlwatch_config.jobs_files)
            and not self.urlwatch_config.config_path.is_dir()
        ):
            self.urlwatch_config.config_path.mkdir(parents=True)
            if not self.urlwatch_config.config_file.is_file():
                self.config_storage.write_default_config(self.urlwatch_config.config_file)
                print(
                    f'A default config has been written to {self.urlwatch_config.config_file}.'
                    f'Use "{__project_name__} --edit-config" to customize it.'
                )

    def load_jobs(self) -> None:
        """Load jobs from the file(s) into self.jobs.

        :raises SystemExit: If job is not found, setting argument to 1.
        """
        if any(f.is_file() for f in self.urlwatch_config.jobs_files):
            jobs = self.jobs_storage.load_secure()
        else:
            print(f'Jobs file not found: {" ,".join((str(file) for file in self.urlwatch_config.jobs_files))}')
            raise SystemExit(1)

        self.jobs = jobs

    def get_new_release_version(self, timeout: float | None = None) -> str | bool:
        """Check PyPi to see if we're running the latest version. Memoized.

        :returns: Empty string if no higher version is available, otherwise the new version number.
        """
        if self._latest_release is not None:
            return self._latest_release

        self._latest_release = get_new_version_number(timeout)
        return self._latest_release

    def run_jobs(self) -> None:
        """Run all jobs."""
        run_jobs(self)

    def close(self) -> None:
        """Finalizer. Create reports ands close snapshots database."""
        self.report.finish(jobs_file=self.jobs_storage.filename)
        # self.ssdb_storage.close()

1	"""The main class.
2
3	For the entry point, see main() in the cli module.
4	"""
5
6	# The code below is subject to the license contained in the LICENSE.md file, which is part of the source code.
7
8	from __future__ import annotations	15✔
9
10	import logging	15✔
11	from typing import TYPE_CHECKING	15✔
12
13	from webchanges import __project_name__	15✔
14	from webchanges.handler import Report	15✔
15	from webchanges.util import get_new_version_number	15✔
16	from webchanges.worker import run_jobs	15✔
17
18	if TYPE_CHECKING:
19	from webchanges.config import CommandConfig
20	from webchanges.jobs import JobBase
21	from webchanges.storage import SsdbStorage, YamlConfigStorage, YamlJobsStorage
22
23
24	logger = logging.getLogger(__name__)	15✔
25
26
27	class Urlwatch:	15✔
28	"""The main class."""
29
30	config_storage: YamlConfigStorage	15✔
31	ssdb_storage: SsdbStorage	15✔
32	jobs_storage: YamlJobsStorage	15✔
33	jobs: list[JobBase]	15✔
34	report: Report	15✔
35
36	def __init__(	15✔
37	self,
38	urlwatch_config: CommandConfig,
39	config_storage: YamlConfigStorage,
40	ssdb_storage: SsdbStorage,
41	jobs_storage: YamlJobsStorage,
42	) -> None:
43	""":param urlwatch_config: The CommandConfig object containing the program run information.
44	:param config_storage: The YamlConfigStorage object containing the configuration information.
45	:param ssdb_storage: The CacheStorage object containing snapshot database information
46	:param jobs_storage: The YamlJobsStorage object containing information about the jobs.
47	"""
48	self.urlwatch_config = urlwatch_config	15✔
49
50	self.config_storage = config_storage	15✔
51	self.ssdb_storage = ssdb_storage	15✔
52	self.jobs_storage = jobs_storage	15✔
53
54	self.report = Report(self)	15✔
55	self.jobs: list[JobBase] = []	15✔
56
57	self._latest_release: str \| bool \| None = None	15✔
58
59	self.check_directories()	15✔
60
61	if not self.urlwatch_config.edit:	15!
62	self.load_jobs()	15✔
63
64	def check_directories(self) -> None:	15✔
65	"""Check whether the configuration and jobs files directories exist. Create them and write default configuration
66	files if one not found.
67	"""
68	if (	15!
69	not (self.urlwatch_config.config_file and self.urlwatch_config.jobs_files)
70	and not self.urlwatch_config.config_path.is_dir()
71	):
72	self.urlwatch_config.config_path.mkdir(parents=True)	×
73	if not self.urlwatch_config.config_file.is_file():	×
74	self.config_storage.write_default_config(self.urlwatch_config.config_file)	×
75	print(	×
76	f'A default config has been written to {self.urlwatch_config.config_file}.'
77	f'Use "{__project_name__} --edit-config" to customize it.'
78	)
79
80	def load_jobs(self) -> None:	15✔
81	"""Load jobs from the file(s) into self.jobs.
82
83	:raises SystemExit: If job is not found, setting argument to 1.
84	"""
85	if any(f.is_file() for f in self.urlwatch_config.jobs_files):	15!
86	jobs = self.jobs_storage.load_secure()	15✔
87	else:
88	print(f'Jobs file not found: {" ,".join((str(file) for file in self.urlwatch_config.jobs_files))}')	×
89	raise SystemExit(1)	×
90
91	self.jobs = jobs	15✔
92
93	def get_new_release_version(self, timeout: float \| None = None) -> str \| bool:	15✔
94	"""Check PyPi to see if we're running the latest version. Memoized.
95
96	:returns: Empty string if no higher version is available, otherwise the new version number.
97	"""
98	if self._latest_release is not None:	6!
99	return self._latest_release	×
100
101	self._latest_release = get_new_version_number(timeout)	6✔
102	return self._latest_release	6✔
103
104	def run_jobs(self) -> None:	15✔
105	"""Run all jobs."""
106	run_jobs(self)	15✔
107
108	def close(self) -> None:	15✔
109	"""Finalizer. Create reports ands close snapshots database."""
110	self.report.finish(jobs_file=self.jobs_storage.filename)	15✔
111	# self.ssdb_storage.close()

mborsetti / webchanges / 21856489627

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous