36151586431

Committed 25 Sep 2026 03:03PM UTC coverage: 90.78% (-0.1%) from 90.909%

Build # 36151586431

Build Type

Pull #1397

github

Committed by

web-flow

Commit Message

Merge a074ec3d5 into dc3b66b05

Pull Request Pull Request #1397: Add type annotations and fix mypy errors in the codebase

Coverage Stats

312 of 332 new or added lines in 33 files covered. (93.98%)

1 existing line in 1 file now uncovered.

3761 of 4143 relevant lines covered (90.78%)

17.65 hits per line

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

64.58

/packages/python-ta/src/python_ta/check/watch.py

"""Module to watch files for modifications and trigger PythonTA checks automatically."""

import logging
import os
import time
from typing import Any, Optional, Union, cast

from pylint.lint import PyLinter
from pylint.reporters import BaseReporter

# ``watchdog`` is an optional extra. If someone imports this module without
# installing the extra, we print an error message before raising the error.
try:
    from watchdog.events import FileSystemEventHandler
    from watchdog.observers import Observer
except ImportError:  # pragma: no cover - impossible to trigger in CI since extra is installed
    print("ERROR: watch mode requires python-ta[watchdog] to be installed.")
    raise

from .helpers import check_file, upload_linter_results


class FileChangeHandler(FileSystemEventHandler):
    """Internal class to handle file modifications."""

    def __init__(
        self,
        files_to_watch: set,
        linter: PyLinter,
        local_config: Union[dict[str, Any], str],
        load_default_config: bool,
        autoformat: Optional[bool],
        level: str,
        f_paths: list[str],
    ) -> None:
        self.files_to_watch = set(files_to_watch)
        self.linter = linter
        self.local_config = local_config
        self.load_default_config = load_default_config
        self.autoformat = autoformat
        self.level = level
        self.f_paths = f_paths

    def on_modified(self, event) -> None:
        """Trigger the callback when a watched file is modified."""
        if event.src_path not in self.files_to_watch:
            return

        logging.info(f"File modified: {event.src_path}, re-running checks...")

        current_reporter = self.linter.reporter
        if event.src_path in current_reporter.messages:
            del current_reporter.messages[event.src_path]

        _, self.linter = check_file(
            file_py=event.src_path,
            local_config=self.local_config,
            load_default_config=self.load_default_config,
            autoformat=self.autoformat,
            is_any_file_checked=True,
            current_reporter=cast(BaseReporter, current_reporter),
            f_paths=[],
        )
        current_reporter = self.linter.reporter
        cast(Any, current_reporter).print_messages(self.level)
        self.linter.generate_reports()
        upload_linter_results(
            self.linter, cast(BaseReporter, current_reporter), self.f_paths, self.local_config
        )


def watch_files(
    file_paths: set,
    level: str,
    local_config: Union[dict[str, Any], str],
    load_default_config: bool,
    autoformat: Optional[bool],
    linter: PyLinter,
    f_paths: list[str],
) -> None:
    """Watch a list of files for modifications and trigger a callback when changes occur."""
    directories_to_watch = {os.path.dirname(file) for file in file_paths}
    event_handler = FileChangeHandler(
        files_to_watch=file_paths,
        linter=linter,
        local_config=local_config,
        load_default_config=load_default_config,
        autoformat=autoformat,
        level=level,
        f_paths=f_paths,
    )
    observer = Observer()
    for directory in directories_to_watch:
        observer.schedule(event_handler, path=directory, recursive=False)
    logging.info("PythonTA is monitoring your files for changes and will re-check after every save")
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        reporter = event_handler.linter.reporter
        cast(Any, reporter).should_close_out = True
        reporter.on_close(event_handler.linter.stats, None)
        observer.stop()

    observer.join()

1	"""Module to watch files for modifications and trigger PythonTA checks automatically."""
2
3	import logging	20 ✔
4	import os	20 ✔
5	import time	20 ✔
6	from typing import Any, Optional, Union, cast	20 ✔
7
8	from pylint.lint import PyLinter	20 ✔
9	from pylint.reporters import BaseReporter	20 ✔
10
11	# ``watchdog`` is an optional extra. If someone imports this module without
12	# installing the extra, we print an error message before raising the error.
13	try:	20 ✔
14	from watchdog.events import FileSystemEventHandler	20 ✔
15	from watchdog.observers import Observer	20 ✔
16	except ImportError: # pragma: no cover - impossible to trigger in CI since extra is installed
17	print("ERROR: watch mode requires python-ta[watchdog] to be installed.")
18	raise
19
20	from .helpers import check_file, upload_linter_results	20 ✔
21
22
23	class FileChangeHandler(FileSystemEventHandler):	20 ✔
24	"""Internal class to handle file modifications."""
25
26	def __init__(	20 ✔
27	self,
28	files_to_watch: set,
29	linter: PyLinter,
30	local_config: Union[dict[str, Any], str],
31	load_default_config: bool,
32	autoformat: Optional[bool],
33	level: str,
34	f_paths: list[str],
35	) -> None:
36	self.files_to_watch = set(files_to_watch)	20 ✔
37	self.linter = linter	20 ✔
38	self.local_config = local_config	20 ✔
39	self.load_default_config = load_default_config	20 ✔
40	self.autoformat = autoformat	20 ✔
41	self.level = level	20 ✔
42	self.f_paths = f_paths	20 ✔
43
44	def on_modified(self, event) -> None:	20 ✔
45	"""Trigger the callback when a watched file is modified."""
46	if event.src_path not in self.files_to_watch:	20 ✔
47	return	20 ✔
48
49	logging.info(f"File modified: {event.src_path}, re-running checks...")	20 ✔
50
51	current_reporter = self.linter.reporter	20 ✔
52	if event.src_path in current_reporter.messages:	20 ✔
53	del current_reporter.messages[event.src_path]	×
54
55	_, self.linter = check_file(	20 ✔
56	file_py=event.src_path,
57	local_config=self.local_config,
58	load_default_config=self.load_default_config,
59	autoformat=self.autoformat,
60	is_any_file_checked=True,
61	current_reporter=cast(BaseReporter, current_reporter),
62	f_paths=[],
63	)
64	current_reporter = self.linter.reporter	20 ✔
65	cast(Any, current_reporter).print_messages(self.level)	20 ✔
66	self.linter.generate_reports()	20 ✔
67	upload_linter_results(	20 ✔
68	self.linter, cast(BaseReporter, current_reporter), self.f_paths, self.local_config
69	)
70
71
72	def watch_files(	20 ✔
73	file_paths: set,
74	level: str,
75	local_config: Union[dict[str, Any], str],
76	load_default_config: bool,
77	autoformat: Optional[bool],
78	linter: PyLinter,
79	f_paths: list[str],
80	) -> None:
81	"""Watch a list of files for modifications and trigger a callback when changes occur."""
82	directories_to_watch = {os.path.dirname(file) for file in file_paths}	×
83	event_handler = FileChangeHandler(	×
84	files_to_watch=file_paths,
85	linter=linter,
86	local_config=local_config,
87	load_default_config=load_default_config,
88	autoformat=autoformat,
89	level=level,
90	f_paths=f_paths,
91	)
92	observer = Observer()	×
93	for directory in directories_to_watch:	×
94	observer.schedule(event_handler, path=directory, recursive=False)	×
95	logging.info("PythonTA is monitoring your files for changes and will re-check after every save")	×
96	observer.start()	×
97
98	try:	×
99	while True:	×
100	time.sleep(1)	×
101	except KeyboardInterrupt:	×
NEW 102	reporter = event_handler.linter.reporter	×
NEW 103	cast(Any, reporter).should_close_out = True	×
NEW 104	reporter.on_close(event_handler.linter.stats, None)	×
UNCOV 105	observer.stop()	×
106
107	observer.join()	×

pyta-uoft / pyta / 36151586431

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