36646375051

Committed 29 Sep 2026 11:39PM UTC coverage: 90.867% (-0.06%) from 90.925%

Build # 36646375051

Build Type

push

github

Committed by

web-flow

Commit Message

Added type annotations and fixed mypy errors in python-ta package (#1397)

Coverage Stats

256 of 269 new or added lines in 33 files covered. (95.17%)

2 existing lines in 2 files now uncovered.

3761 of 4139 relevant lines covered (90.87%)

17.67 hits per line

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

65.96

/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

from pylint.lint import PyLinter

from python_ta.reporters.core import PythonTaReporter

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

pyta-uoft / pyta / 36646375051

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