Complete python-crontab (#9306)

Co-authored-by: Alex Waygood <[email protected]>
Co-authored-by: Jelle Zijlstra <[email protected]>

Author: 3 people

pyrightconfig.stricter.json

@@ -49,7 +49,6 @@
         "stubs/Pillow",
         "stubs/prettytable",
         "stubs/protobuf",
-        "stubs/python-crontab",
         "stubs/google-cloud-ndb",
         "stubs/influxdb-client",
         "stubs/passlib",

stubs/python-crontab/@tests/stubtest_allowlist.txt

@@ -1,2 +1,4 @@
+# Runtime only-hack that doesn't affect typing:
+crontabs.CronTabs.__new__
 # stub does not have *args argument "args", but function doesn't actually accept positional args
 crontab.CronTab.remove_all

stubs/python-crontab/crontab.pyi

@@ -1,15 +1,19 @@
 import re
 import subprocess
 from _typeshed import Incomplete, Self
+from builtins import range as _range
 from collections import OrderedDict
-from collections.abc import Callable, Generator, Iterator
+from collections.abc import Callable, Generator, Iterable, Iterator
+from datetime import datetime
 from logging import Logger
 from types import TracebackType
 from typing import Any
-from typing_extensions import SupportsIndex
+from typing_extensions import SupportsIndex, TypeAlias
 
 from cronlog import CronLog
 
+_User: TypeAlias = str | bool | None
+
 __pkgname__: str
 ITEMREX: re.Pattern[str]
 SPECREX: re.Pattern[str]
@@ -28,18 +32,19 @@ CRON_COMMAND: str
 SHELL: str
 current_user: Callable[[], str | None]
 
-def open_pipe(cmd: str, *args: str, **flags) -> subprocess.Popen[Any]: ...
+def open_pipe(cmd: str, *args: str, **flags: str) -> subprocess.Popen[Any]: ...
 
 class CronTab:
-    lines: Incomplete
-    crons: list[CronItem]
+    lines: list[str | CronItem] | None
+    crons: list[CronItem] | None
     filen: str | None
-    cron_command: Incomplete
-    env: OrderedVariableList
+    cron_command: str
+    env: OrderedVariableList | None
     root: bool
     intab: str | None
+    tabfile: str | None
     def __init__(
-        self, user: bool | str | None = ..., tab: str | None = ..., tabfile: str | None = ..., log: str | None = ...
+        self, user: _User = ..., tab: str | None = ..., tabfile: str | None = ..., log: CronLog | str | None = ...
     ) -> None: ...
     def __enter__(self: Self) -> Self: ...
     def __exit__(
@@ -48,7 +53,7 @@ class CronTab:
     @property
     def log(self) -> CronLog: ...
     @property
-    def user(self) -> str | None: ...
+    def user(self) -> _User: ...
     @property
     def user_opt(self) -> dict[str, str]: ...
     def read(self, filename: str | None = ...) -> None: ...
@@ -59,11 +64,14 @@ class CronTab:
         read: bool = ...,
         before: str | re.Pattern[str] | list[CronItem] | tuple[CronItem, ...] | Generator[CronItem, Any, Any] | None = ...,
     ) -> None: ...
-    def write(self, filename: str | None = ..., user: bool | str | None = ..., errors: bool = ...) -> None: ...
+    def write(self, filename: str | None = ..., user: _User = ..., errors: bool = ...) -> None: ...
     def write_to_user(self, user: bool | str = ...) -> None: ...
-    def run_pending(self, **kwargs) -> Generator[Incomplete, None, None]: ...
-    def run_scheduler(self, timeout: int = ..., **kwargs) -> Generator[Incomplete, None, None]: ...
-    def render(self, errors: bool = ..., specials: bool = ...) -> str: ...
+    # Usually `kwargs` are just `now: datetime | None`, but technically this can
+    # work for `CronItem` subclasses, which might define other kwargs.
+    def run_pending(self, **kwargs: Any) -> Iterator[str]: ...
+    # There are two known kwargs and others are unused:
+    def run_scheduler(self, timeout: int = ..., *, warp: object = ..., cadence: int = ..., **kwargs: object) -> Iterator[str]: ...
+    def render(self, errors: bool = ..., specials: bool | None = ...) -> str: ...
     def new(
         self,
         command: str = ...,
@@ -72,97 +80,104 @@ class CronTab:
         pre_comment: bool = ...,
         before: str | re.Pattern[str] | list[CronItem] | tuple[CronItem, ...] | Generator[CronItem, Any, Any] | None = ...,
     ) -> CronItem: ...
-    def find_command(self, command: str | re.Pattern[str]) -> Generator[CronItem, None, None]: ...
-    def find_comment(self, comment: str | re.Pattern[str]) -> Generator[CronItem, None, None]: ...
-    def find_time(self, *args) -> Generator[CronItem, None, None]: ...
+    def find_command(self, command: str | re.Pattern[str]) -> Iterator[CronItem]: ...
+    def find_comment(self, comment: str | re.Pattern[str]) -> Iterator[CronItem]: ...
+    def find_time(self, *args: Any) -> Iterator[CronItem]: ...
     @property
-    def commands(self) -> Generator[Incomplete, None, None]: ...
+    def commands(self) -> Iterator[str]: ...
     @property
-    def comments(self) -> Generator[Incomplete, None, None]: ...
-    def remove_all(self, *, command: str | re.Pattern[str] = ..., comment: str | re.Pattern[str] = ..., time=...) -> int: ...
-    def remove(self, *items: CronItem | list[CronItem] | tuple[CronItem, ...] | Generator[CronItem, Any, Any]) -> int: ...
+    def comments(self) -> Iterator[str]: ...
+    # You cannot actually pass `*args`, it will raise an exception,
+    # also known kwargs are added:
+    def remove_all(
+        self, *, command: str | re.Pattern[str] = ..., comment: str | re.Pattern[str] = ..., time: Any = ..., **kwargs: object
+    ) -> int: ...
+    def remove(self, *items: CronItem | Iterable[CronItem]) -> int: ...
     def __iter__(self) -> Iterator[CronItem]: ...
     def __getitem__(self, i: SupportsIndex) -> CronItem: ...
     def __len__(self) -> int: ...
 
 class CronItem:
-    cron: Incomplete
-    user: str | None
+    cron: CronTab | None
+    user: _User
     valid: bool
     enabled: bool
     special: bool
-    comment: Incomplete
-    command: Incomplete
-    last_run: Incomplete
-    env: Incomplete
+    comment: str
+    command: str | None
+    last_run: datetime | None
+    env: OrderedVariableList
     pre_comment: bool
-    marker: Incomplete
-    stdin: Incomplete
-    slices: Incomplete
-    def __init__(self, command: str = ..., comment: str = ..., user: str | None = ..., pre_comment: bool = ...) -> None: ...
+    marker: str | None
+    stdin: str | None
+    slices: CronSlices
+    def __init__(self, command: str = ..., comment: str = ..., user: _User = ..., pre_comment: bool = ...) -> None: ...
     def __hash__(self) -> int: ...
     def __eq__(self, other: object) -> bool: ...
     @classmethod
     def from_line(cls: type[Self], line: str, user: str | None = ..., cron: Incomplete | None = ...) -> Self: ...
     def delete(self) -> None: ...
     def set_command(self, cmd: str, parse_stdin: bool = ...) -> None: ...
     def set_comment(self, cmt: str, pre_comment: bool = ...) -> None: ...
-    def parse(self, line) -> None: ...
+    def parse(self, line: str) -> None: ...
     def enable(self, enabled: bool = ...) -> bool: ...
     def is_enabled(self) -> bool: ...
     def is_valid(self) -> bool: ...
     def render(self, specials: bool = ...) -> str: ...
-    def every_reboot(self): ...
-    def every(self, unit: int = ...): ...
-    def setall(self, *args: Any): ...
-    def clear(self): ...
+    def every_reboot(self) -> None: ...
+    def every(self, unit: int = ...) -> Every: ...
+    def setall(self, *args: Any) -> None: ...
+    def clear(self) -> None: ...
     def frequency(self, year: int | None = ...) -> int: ...
     def frequency_per_year(self, year: int | None = ...) -> int: ...
     def frequency_per_day(self) -> int: ...
     def frequency_per_hour(self) -> int: ...
-    def run_pending(self, now: Incomplete | None = ...): ...
-    def run(self): ...
-    def schedule(self, date_from: Incomplete | None = ...): ...
-    def description(self, **kw: Any): ...
+    def run_pending(self, now: datetime | None = ...) -> int | str: ...
+    def run(self) -> str: ...
+    # TODO: use types from `croniter` module here:
+    def schedule(self, date_from: datetime | None = ...) -> Incomplete: ...
+    # TODO: use types from `cron_descriptor` here:
+    def description(self, **kw: Incomplete) -> Incomplete: ...
     @property
-    def log(self): ...
+    def log(self) -> CronLog: ...
     @property
-    def minute(self) -> CronSlice: ...
+    def minute(self) -> int | str: ...
     @property
-    def minutes(self) -> CronSlice: ...
+    def minutes(self) -> int | str: ...
     @property
-    def hour(self) -> CronSlice: ...
+    def hour(self) -> int | str: ...
     @property
-    def hours(self) -> CronSlice: ...
+    def hours(self) -> int | str: ...
     @property
-    def day(self) -> CronSlice: ...
+    def day(self) -> int | str: ...
     @property
-    def dom(self) -> CronSlice: ...
+    def dom(self) -> int | str: ...
     @property
-    def month(self) -> CronSlice: ...
+    def month(self) -> int | str: ...
     @property
-    def months(self) -> CronSlice: ...
+    def months(self) -> int | str: ...
     @property
-    def dow(self) -> CronSlice: ...
+    def dow(self) -> int | str: ...
     def __len__(self) -> int: ...
-    def __getitem__(self, key: SupportsIndex) -> CronSlice: ...
-    def __lt__(self, value) -> bool: ...
-    def __gt__(self, value) -> bool: ...
+    def __getitem__(self, key: int | str) -> int | str: ...
+    def __lt__(self, value: object) -> bool: ...
+    def __gt__(self, value: object) -> bool: ...
 
 class Every:
-    slices: Incomplete
-    unit: Incomplete
-    def __init__(self, item, units) -> None: ...
+    slices: CronSlices
+    unit: int
+    # TODO: add generated attributes
+    def __init__(self, item: CronSlices, units: int) -> None: ...
     def set_attr(self, target: int) -> Callable[[], None]: ...
     def year(self) -> None: ...
 
 class CronSlices(list[CronSlice]):
-    special: Incomplete
+    special: bool | None
     def __init__(self, *args: Any) -> None: ...
     def is_self_valid(self, *args: Any) -> bool: ...
     @classmethod
     def is_valid(cls, *args: Any) -> bool: ...
-    def setall(self, *slices) -> None: ...
+    def setall(self, *slices: str) -> None: ...
     def clean_render(self) -> str: ...
     def render(self, specials: bool = ...) -> str: ...
     def clear(self) -> None: ...
@@ -175,64 +190,72 @@ class CronSlices(list[CronSlice]):
 class SundayError(KeyError): ...
 
 class Also:
-    obj: Incomplete
-    def __init__(self, obj) -> None: ...
-    def every(self, *a): ...
-    def on(self, *a): ...
-    def during(self, *a): ...
+    obj: CronSlice
+    def __init__(self, obj: CronSlice) -> None: ...
+    # These method actually use `*args`, but pass them to `CronSlice` methods,
+    # this is why they are typed as `Any`.
+    def every(self, *a: Any) -> _Part: ...
+    def on(self, *a: Any) -> list[_Part]: ...
+    def during(self, *a: Any) -> _Part: ...
+
+_Part: TypeAlias = int | CronValue | CronRange
 
 class CronSlice:
-    min: Incomplete
-    max: Incomplete
-    name: Incomplete
-    enum: Incomplete
-    parts: Incomplete
-    def __init__(self, info, value: Incomplete | None = ...) -> None: ...
+    min: int | None
+    max: int | None
+    name: str | None
+    enum: list[str | None] | None
+    parts: list[_Part]
+    def __init__(self, info: int | dict[str, Any], value: str | None = ...) -> None: ...
     def __hash__(self) -> int: ...
-    def parse(self, value) -> None: ...
-    def render(self, resolve: bool = ..., specials: bool = ...): ...
+    def parse(self, value: str | None) -> None: ...
+    def render(self, resolve: bool = ..., specials: bool = ...) -> str: ...
     def __eq__(self, arg: object) -> bool: ...
-    def every(self, n_value, also: bool = ...): ...
-    def on(self, *n_value, **opts): ...
-    def during(self, vfrom, vto, also: bool = ...): ...
+    def every(self, n_value: int, also: bool = ...) -> _Part: ...
+    # The only known kwarg, others are unused,
+    # `*args`` are passed to `parse_value`, so they are `Any`
+    def on(self, *n_value: Any, also: bool = ...) -> list[_Part]: ...
+    def during(self, vfrom: int | str, vto: int | str, also: bool = ...) -> _Part: ...
     @property
-    def also(self): ...
+    def also(self) -> Also: ...
     def clear(self) -> None: ...
-    def get_range(self, *vrange): ...
-    def __iter__(self): ...
+    def get_range(self, *vrange: int | str | CronValue) -> list[int | CronRange]: ...
+    def __iter__(self) -> Iterator[int]: ...
     def __len__(self) -> int: ...
-    def parse_value(self, val, sunday: Incomplete | None = ...): ...
+    def parse_value(self, val: str, sunday: int | None = ...) -> int | CronValue: ...
 
-def get_cronvalue(value, enums): ...
+def get_cronvalue(value: int, enums: list[str]) -> int | CronValue: ...
 
 class CronValue:
-    text: Incomplete
-    value: Incomplete
-    def __init__(self, value, enums) -> None: ...
-    def __lt__(self, value): ...
+    text: str
+    value: int
+    def __init__(self, value: str, enums: list[str]) -> None: ...
+    def __lt__(self, value: object) -> bool: ...
     def __int__(self) -> int: ...
 
 class CronRange:
-    dangling: Incomplete
-    slice: Incomplete
-    cron: Incomplete
+    dangling: int | None
+    slice: str
+    cron: CronTab | None
     seq: int
-    def __init__(self, vslice, *vrange) -> None: ...
-    vfrom: Incomplete
-    vto: Incomplete
-    def parse(self, value) -> None: ...
+    def __init__(self, vslice: str, *vrange: int | str | CronValue) -> None: ...
+    # Are not set in `__init__`:
+    vfrom: int | CronValue
+    vto: int | CronValue
+    def parse(self, value: str) -> None: ...
     def all(self) -> None: ...
-    def render(self, resolve: bool = ...): ...
-    def range(self): ...
-    def every(self, value) -> None: ...
-    def __lt__(self, value): ...
-    def __gt__(self, value): ...
+    def render(self, resolve: bool = ...) -> str: ...
+    def range(self) -> _range: ...
+    def every(self, value: int | str) -> None: ...
+    def __lt__(self, value: object) -> bool: ...
+    def __gt__(self, value: object) -> bool: ...
     def __int__(self) -> int: ...
 
+# TODO: make generic
 class OrderedVariableList(OrderedDict[Incomplete, Incomplete]):
     job: Incomplete
     def __init__(self, *args: Any, **kw: Any) -> None: ...
     @property
-    def previous(self): ...
-    def all(self): ...
-    def __getitem__(self, key): ...
+    def previous(self) -> Incomplete: ...
+    def all(self: Self) -> Self: ...
+    def __getitem__(self, key: Incomplete) -> Incomplete: ...

stubs/python-crontab/crontabs.pyi

@@ -13,12 +13,11 @@ class SystemTab(list[CronTab]):
 
 class AnaCronTab(list[CronTab]):
     def __init__(self, loc: str, tabs: CronTabs | None = ...) -> None: ...
-    def add(self, loc: str, item: str, anajob) -> CronTab: ...
+    def add(self, loc: str, item: str, anajob: CronTab) -> CronTab: ...
 
 KNOWN_LOCATIONS: list[tuple[UserSpool | SystemTab | AnaCronTab, str]]
 
 class CronTabs(list[UserSpool | SystemTab | AnaCronTab]):
-    def __new__(cls, *args: Any, **kw: Any): ...
     def __init__(self) -> None: ...
     def add(self, cls: type[UserSpool | SystemTab | AnaCronTab], *args: Any) -> None: ...
     @property