14049646952

Committed 25 Mar 2025 01:34AM UTC coverage: 92.557% (-0.5%) from 93.089%

Build # 14049646952

Build Type

push

github

Committed by

web-flow

Commit Message

Modify check_all to re-run checks when watch=True (#1156)

Adds the watchdog library to monitor file system events.

Run Details

192 of 218 new or added lines in 6 files covered. (88.07%)

8 existing lines in 1 file now uncovered.

3345 of 3614 relevant lines covered (92.56%)

17.63 hits per line

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

66.67

/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 pylint.reporters import BaseReporter, MultiReporter
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer

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

pyta-uoft / pyta / 14049646952

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