add cron job

Author: Nanguage

executor/engine/__init__.py

@@ -1,7 +1,7 @@
 from .core import Engine, EngineSetting
 from .job import LocalJob, ThreadJob, ProcessJob
 
-__version__ = '0.2.8'
+__version__ = '0.2.9'
 
 __all__ = [
     'Engine', 'EngineSetting',

executor/engine/core.py

@@ -37,6 +37,10 @@ class EngineSetting:
             if not set, will create a cache directory in
             .executor/{engine.id}.
         print_traceback: Whether to print traceback when job failed.
+        kwargs_inject_key: Key to inject engine to job kwargs.
+            If set, the engine will be injected to job kwargs
+            with the key.
+            default is "__engine__".
     """
     max_thread_jobs: T.Optional[int] = None
     max_process_jobs: T.Optional[int] = None
@@ -45,6 +49,7 @@ class EngineSetting:
     cache_type: T.Literal["diskcache", "none"] = "none"
     cache_path: T.Optional[str] = None
     print_traceback: bool = True
+    kwargs_inject_key: str = "__engine__"
 
 
 @dataclass
@@ -193,6 +198,9 @@ async def submit_async(self, *jobs: Job):
             if job.status == "created":
                 job.engine = self
                 job._status = "pending"
+                func_var_names = job.func.__code__.co_varnames
+                if self.setting.kwargs_inject_key in func_var_names:
+                    job.kwargs[self.setting.kwargs_inject_key] = self
                 self.jobs.add(job)
             else:
                 job.status = "pending"

executor/engine/job/condition.py

@@ -1,5 +1,5 @@
 import typing as T
-from datetime import datetime
+from datetime import datetime, timedelta
 from dataclasses import dataclass
 
 from .utils import JobStatusType
@@ -16,6 +16,12 @@ def satisfy(self, engine: "Engine") -> bool:  # pragma: no cover
         """Check if the condition is satisfied."""
         return True
 
+    def __or__(self, other: "Condition") -> "AnySatisfied":
+        return AnySatisfied(conditions=[self, other])
+
+    def __and__(self, other: "Condition") -> "AllSatisfied":
+        return AllSatisfied(conditions=[self, other])
+
 
 @dataclass
 class AfterAnother(Condition):
@@ -30,8 +36,7 @@ class AfterAnother(Condition):
     job_id: str
     statuses: T.Iterable[JobStatusType] = ("done", "failed", "cancelled")
 
-    def satisfy(self, engine):
-        """Check if the condition is satisfied."""
+    def satisfy(self, engine) -> bool:
         try:
             another = engine.jobs.get_job_by_id(self.job_id)
         except Exception:
@@ -59,8 +64,7 @@ class AfterOthers(Condition):
     statuses: T.Iterable[JobStatusType] = ("done", "failed", "cancelled")
     mode: T.Literal['all', 'any'] = "all"
 
-    def satisfy(self, engine):
-        """Check if the condition is satisfied."""
+    def satisfy(self, engine) -> bool:
         other_job_satisfy = []
         for id_ in self.job_ids:
             try:
@@ -76,24 +80,6 @@ def satisfy(self, engine):
             return any(other_job_satisfy)
 
 
-@dataclass
-class AfterTimepoint(Condition):
-    """Condition that the job is executed after a timepoint.
-
-    Attributes:
-        timepoint: The timepoint.
-    """
-
-    timepoint: datetime
-
-    def satisfy(self, engine):
-        """Check if the condition is satisfied."""
-        if datetime.now() > self.timepoint:
-            return True
-        else:
-            return False
-
-
 @dataclass
 class Combination(Condition):
     """Base class for combination of conditions.
@@ -134,3 +120,217 @@ class AnySatisfied(Combination):
     def satisfy(self, engine):
         """Check if the condition is satisfied."""
         return any([c.satisfy(engine) for c in self.conditions])
+
+
+def _parse_clock_str(time_str: str) -> T.Tuple[int, int, int]:
+    hour: int
+    minute: int
+    second: int
+    if time_str.count(":") == 1:
+        hour, minute = time_str.split(":")
+        hour, minute = int(hour), int(minute)
+        second = 0
+    elif time_str.count(":") == 2:
+        hour, minute, second = time_str.split(":")
+        hour, minute, second = int(hour), int(minute), int(second)
+    else:
+        hour = int(time_str)
+        minute = 0
+        second = 0
+    return hour, minute, second
+
+
+def _parse_weekday_str(weekday_str: str) -> int:
+    if weekday_str.lower() in ("monday", "mon"):
+        return 0
+    elif weekday_str.lower() in ("tuesday", "tue"):
+        return 1
+    elif weekday_str.lower() in ("wednesday", "wed"):
+        return 2
+    elif weekday_str.lower() in ("thursday", "thu"):
+        return 3
+    elif weekday_str.lower() in ("friday", "fri"):
+        return 4
+    elif weekday_str.lower() in ("saturday", "sat"):
+        return 5
+    elif weekday_str.lower() in ("sunday", "sun"):
+        return 6
+    raise ValueError(f"Invalid weekday string: {weekday_str}")
+
+
+def _parse_period_str(period_str: str) -> timedelta:
+    if period_str.endswith("d"):
+        return timedelta(days=float(period_str[:-1]))
+    elif period_str.endswith("h"):
+        return timedelta(hours=float(period_str[:-1]))
+    elif period_str.endswith("m"):
+        return timedelta(minutes=float(period_str[:-1]))
+    elif period_str.endswith("s"):
+        return timedelta(seconds=float(period_str[:-1]))
+    raise ValueError(f"Invalid period string: {period_str}")
+
+
+@dataclass
+class TimeCondition(Condition):
+    pass
+
+
+@dataclass
+class EveryPeriod(TimeCondition):
+    """Every period.
+
+    Attributes:
+        period_str: The period string.
+            format: "1d", "1h", "1m", "1s"
+        last_submitted_at: The last submitted time.
+        immediate: Whether to submit the job immediately.
+    """
+    period_str: str
+    last_submitted_at: T.Optional[datetime] = None
+    immediate: bool = True
+
+    def satisfy(self, _) -> bool:
+        period = _parse_period_str(self.period_str)
+        res: bool
+        if self.last_submitted_at is None:
+            self.last_submitted_at = datetime.now()
+            res = self.immediate
+        else:
+            res = (datetime.now() - self.last_submitted_at >= period)
+        if res:
+            self.last_submitted_at = datetime.now()
+        return res
+
+
+@dataclass
+class AfterClock(TimeCondition):
+    """After a specific clock.
+
+    Attributes:
+        time_str: The time string.
+            format: "12:00", "12:00:00"
+    """
+    time_str: str
+
+    def satisfy(self, _) -> bool:
+        hour, minute, second = _parse_clock_str(self.time_str)
+        now = datetime.now()
+        res = (
+            (now.hour >= hour) &
+            (now.minute >= minute) &
+            (now.second >= second)
+        )
+        return res
+
+
+@dataclass
+class BeforeClock(TimeCondition):
+    """Before a specific clock.
+
+    Attributes:
+        time_str: The time string.
+            format: "12:00", "12:00:00"
+    """
+    time_str: str
+
+    def satisfy(self, _) -> bool:
+        hour, minute, second = _parse_clock_str(self.time_str)
+        now = datetime.now()
+        res = (
+            (now.hour <= hour) &
+            (now.minute <= minute) &
+            (now.second <= second)
+        )
+        return res
+
+
+@dataclass
+class AfterWeekday(TimeCondition):
+    """After a specific weekday.
+
+    Attributes:
+        weekday_str: The weekday string.
+            format: "monday", "mon", "tuesday", "tue", "wednesday", "wed",
+                "thursday", "thu", "friday", "fri", "saturday", "sat",
+                "sunday", "sun"
+    """
+    weekday_str: str
+
+    def satisfy(self, _) -> bool:
+        weekday = _parse_weekday_str(self.weekday_str)
+        now = datetime.now()
+        res = (now.weekday() >= weekday)
+        return res
+
+
+@dataclass
+class BeforeWeekday(TimeCondition):
+    """Before a specific weekday.
+
+    Attributes:
+        weekday_str: The weekday string.
+            format: "monday", "mon", "tuesday", "tue", "wednesday", "wed",
+                "thursday", "thu", "friday", "fri", "saturday", "sat",
+                "sunday", "sun"
+    """
+    weekday_str: str
+
+    def satisfy(self, _) -> bool:
+        weekday = _parse_weekday_str(self.weekday_str)
+        now = datetime.now()
+        res = (now.weekday() <= weekday)
+        return res
+
+
+_valid_timepoint_fields = ("year", "month", "day", "hour", "minute", "second")
+
+
+@dataclass
+class AfterTimepoint(TimeCondition):
+    """After a timepoint.
+
+    Attributes:
+        timepoint: The timepoint.
+        compare_fields: The fields to compare.
+            Fields: "year", "month", "day", "hour", "minute", "second"
+    """
+
+    timepoint: datetime
+    compare_fields: T.Optional[T.List[str]] = None
+
+    def satisfy(self, _) -> bool:
+        if self.compare_fields is None:
+            return datetime.now() > self.timepoint
+        else:
+            for field in self.compare_fields:
+                if field not in _valid_timepoint_fields:
+                    raise ValueError(f"Invalid field: {field}")
+                if getattr(datetime.now(), field) >= \
+                   getattr(self.timepoint, field):
+                    return False
+            return True
+
+
+@dataclass
+class BeforeTimepoint(TimeCondition):
+    """Before a timepoint.
+
+    Attributes:
+        timepoint: The timepoint.
+        compare_fields: The fields to compare.
+            Fields: "year", "month", "day", "hour", "minute", "second"
+    """
+    timepoint: datetime
+    compare_fields: T.Optional[T.List[str]] = None
+
+    def satisfy(self, _) -> bool:
+        if self.compare_fields is None:
+            return datetime.now() < self.timepoint
+        else:
+            for field in self.compare_fields:
+                if field not in _valid_timepoint_fields:
+                    raise ValueError(f"Invalid field: {field}")
+                if getattr(datetime.now(), field) <= \
+                   getattr(self.timepoint, field):
+                    return False
+            return True

executor/engine/job/extend/cron.py

@@ -0,0 +1,71 @@
+import typing as T
+from datetime import datetime
+
+from .. import Job
+from ..condition import (
+    TimeCondition,
+    AllSatisfied, AfterTimepoint, BeforeTimepoint,
+    EveryPeriod, BeforeClock, AfterClock,
+    AfterWeekday, BeforeWeekday
+)
+from .sentinel import SentinelJob
+
+
+every = EveryPeriod
+daily = EveryPeriod("1d")
+hourly = EveryPeriod("1h")
+monthly = EveryPeriod("1m")
+
+before_clock = BeforeClock
+after_clock = AfterClock
+
+after_weekday = AfterWeekday
+before_weekday = BeforeWeekday
+
+after_timepoint = AfterTimepoint
+before_timepoint = BeforeTimepoint
+
+
+def between_clock(start: str, end: str) -> AllSatisfied:
+    return AllSatisfied([BeforeClock(end), AfterClock(start)])
+
+
+def between_weekday(start: str, end: str) -> AllSatisfied:
+    return AllSatisfied([BeforeWeekday(end), AfterWeekday(start)])
+
+
+def between_timepoint(start: datetime, end: datetime) -> AllSatisfied:
+    return AllSatisfied([BeforeTimepoint(end), AfterTimepoint(start)])
+
+
+def CronJob(
+        func: T.Callable,
+        time_condition: TimeCondition,
+        job_type: T.Union[str, T.Type[Job]] = "process",
+        time_delta: float = 0.01,
+        sentinel_attrs: T.Optional[dict] = None,
+        **attrs
+        ):
+    """Submit a job periodically.
+
+    Args:
+        func: The function to be executed.
+        time_period: The time period.
+        job_type: The type of the job.
+        time_delta: The time delta between each
+            check of the sentinel condition.
+        sentinel_attrs: The attributes of the sentinel job.
+        **attrs: The attributes of the job.
+    """
+    sentinel_attrs = sentinel_attrs or {}
+    if "name" not in sentinel_attrs:
+        sentinel_attrs["name"] = f"cron-sentinel-{func.__name__}"
+    sentinel_job = SentinelJob(
+        func,
+        time_condition,
+        job_type,
+        time_delta,
+        sentinel_attrs,
+        **attrs
+    )
+    return sentinel_job