Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ omit = ["*/__main__.py"]

[tool.ruff]
line-length = 120
target-version = "py38"
target-version = "py39"
lint.mccabe = { max-complexity = 14 }
lint.extend-select = ["Q", "RUF100", "C90", "UP", "I"]
lint.flake8-quotes = { inline-quotes = "single", multiline-quotes = "double" }
Expand Down
4 changes: 2 additions & 2 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
from pathlib import Path
from threading import Thread
from time import sleep, time
from typing import TYPE_CHECKING, Any, List, Set, Tuple
from typing import TYPE_CHECKING, Any

import pytest

Expand Down Expand Up @@ -61,7 +61,7 @@ def start(path: Path):
t.join()


ChangesType = List[Set[Tuple[int, str]]]
ChangesType = list[set[tuple[int, str]]]


class MockRustNotify:
Expand Down
6 changes: 3 additions & 3 deletions watchfiles/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import sys
from pathlib import Path
from textwrap import dedent
from typing import Any, Callable, List, Optional, Tuple, Union, cast
from typing import Any, Callable, Optional, Union, cast

from . import Change
from .filters import BaseFilter, DefaultFilter, PythonFilter
Expand Down Expand Up @@ -196,8 +196,8 @@ def import_exit(function_path: str) -> Any:

def build_filter(
filter_name: str, ignore_paths_str: Optional[str]
) -> Tuple[Union[None, DefaultFilter, Callable[[Change, str], bool]], str]:
ignore_paths: List[Path] = []
) -> tuple[Union[None, DefaultFilter, Callable[[Change, str], bool]], str]:
ignore_paths: list[Path] = []
if ignore_paths_str:
ignore_paths = [Path(p).resolve() for p in ignore_paths_str.split(',')]

Expand Down
3 changes: 2 additions & 1 deletion watchfiles/filters.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
import logging
import os
import re
from collections.abc import Sequence
from pathlib import Path
from typing import TYPE_CHECKING, Optional, Sequence, Union
from typing import TYPE_CHECKING, Optional, Union

__all__ = 'BaseFilter', 'DefaultFilter', 'PythonFilter'
logger = logging.getLogger('watchfiles.watcher')
Expand Down
15 changes: 8 additions & 7 deletions watchfiles/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,10 @@
import os
import sys
import warnings
from collections.abc import AsyncGenerator, Generator
from enum import IntEnum
from pathlib import Path
from typing import TYPE_CHECKING, AsyncGenerator, Callable, Generator, Optional, Set, Tuple, Union
from typing import TYPE_CHECKING, Callable, Optional, Union

import anyio

Expand All @@ -31,7 +32,7 @@ def raw_str(self) -> str:
return self.name


FileChange = Tuple[Change, str]
FileChange = tuple[Change, str]
"""
A tuple representing a file change, first element is a [`Change`][watchfiles.Change] member, second is the path
of the file or directory that changed.
Expand Down Expand Up @@ -63,7 +64,7 @@ def watch(
poll_delay_ms: int = 300,
recursive: bool = True,
ignore_permission_denied: Optional[bool] = None,
) -> Generator[Set[FileChange], None, None]:
) -> Generator[set[FileChange], None, None]:
"""
Watch one or more paths and yield a set of changes whenever files change.

Expand Down Expand Up @@ -164,7 +165,7 @@ async def awatch( # C901
poll_delay_ms: int = 300,
recursive: bool = True,
ignore_permission_denied: Optional[bool] = None,
) -> AsyncGenerator[Set[FileChange], None]:
) -> AsyncGenerator[set[FileChange], None]:
"""
Asynchronous equivalent of [`watch`][watchfiles.watch] using threads to wait for changes.
Arguments match those of [`watch`][watchfiles.watch] except `stop_event`.
Expand Down Expand Up @@ -289,16 +290,16 @@ async def stop_soon():


def _prep_changes(
raw_changes: Set[Tuple[int, str]], watch_filter: Optional[Callable[[Change, str], bool]]
) -> Set[FileChange]:
raw_changes: set[tuple[int, str]], watch_filter: Optional[Callable[[Change, str], bool]]
) -> set[FileChange]:
# if we wanted to be really snazzy, we could move this into rust
changes = {(Change(change), path) for change, path in raw_changes}
if watch_filter:
changes = {c for c in changes if watch_filter(c[0], c[1])}
return changes


def _log_changes(changes: Set[FileChange]) -> None:
def _log_changes(changes: set[FileChange]) -> None:
if logger.isEnabledFor(logging.INFO): # pragma: no branch
count = len(changes)
plural = '' if count == 1 else 's'
Expand Down
25 changes: 13 additions & 12 deletions watchfiles/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,13 @@
import signal
import subprocess
import sys
from collections.abc import Generator
from importlib import import_module
from multiprocessing import get_context
from multiprocessing.context import SpawnProcess
from pathlib import Path
from time import sleep
from typing import TYPE_CHECKING, Any, Callable, Dict, Generator, List, Optional, Set, Tuple, Union
from typing import TYPE_CHECKING, Any, Callable, Optional, Union

import anyio

Expand All @@ -29,10 +30,10 @@
def run_process(
*paths: Union[Path, str],
target: Union[str, Callable[..., Any]],
args: Tuple[Any, ...] = (),
kwargs: Optional[Dict[str, Any]] = None,
args: tuple[Any, ...] = (),
kwargs: Optional[dict[str, Any]] = None,
target_type: "Literal['function', 'command', 'auto']" = 'auto',
callback: Optional[Callable[[Set[FileChange]], None]] = None,
callback: Optional[Callable[[set[FileChange]], None]] = None,
watch_filter: Optional[Callable[[Change, str], bool]] = DefaultFilter(),
grace_period: float = 0,
debounce: int = 1_600,
Expand Down Expand Up @@ -157,10 +158,10 @@ def foobar(a, b, c):
async def arun_process(
*paths: Union[Path, str],
target: Union[str, Callable[..., Any]],
args: Tuple[Any, ...] = (),
kwargs: Optional[Dict[str, Any]] = None,
args: tuple[Any, ...] = (),
kwargs: Optional[dict[str, Any]] = None,
target_type: "Literal['function', 'command', 'auto']" = 'auto',
callback: Optional[Callable[[Set[FileChange]], Any]] = None,
callback: Optional[Callable[[set[FileChange]], Any]] = None,
watch_filter: Optional[Callable[[Change, str], bool]] = DefaultFilter(),
grace_period: float = 0,
debounce: int = 1_600,
Expand Down Expand Up @@ -239,7 +240,7 @@ async def main():
spawn_context = get_context('spawn')


def split_cmd(cmd: str) -> List[str]:
def split_cmd(cmd: str) -> list[str]:
import platform

posix = platform.uname().system.lower() != 'windows'
Expand All @@ -249,9 +250,9 @@ def split_cmd(cmd: str) -> List[str]:
def start_process(
target: Union[str, Callable[..., Any]],
target_type: "Literal['function', 'command']",
args: Tuple[Any, ...],
kwargs: Optional[Dict[str, Any]],
changes: Optional[Set[FileChange]] = None,
args: tuple[Any, ...],
kwargs: Optional[dict[str, Any]],
changes: Optional[set[FileChange]] = None,
) -> 'CombinedProcess':
if changes is None:
changes_env_var = '[]'
Expand Down Expand Up @@ -368,7 +369,7 @@ def exitcode(self) -> Optional[int]:
return self._p.returncode


def run_function(function: str, tty_path: Optional[str], args: Tuple[Any, ...], kwargs: Dict[str, Any]) -> None:
def run_function(function: str, tty_path: Optional[str], args: tuple[Any, ...], kwargs: dict[str, Any]) -> None:
with set_tty(tty_path):
func = import_string(function)
func(*args, **kwargs)
Expand Down
Loading