ISSUE #306

* Add `CronTab` class.

Author: Sergio García Prado

packages/core/minos-microservice-networks/minos/networks/__init__.py

@@ -117,6 +117,7 @@
     Router,
 )
 from .scheduling import (
+    CronTab,
     PeriodicTask,
     PeriodicTaskScheduler,
     PeriodicTaskSchedulerPort,

packages/core/minos-microservice-networks/minos/networks/decorators/definitions/periodic.py

@@ -1,36 +1,48 @@
+from __future__ import (
+    annotations,
+)
+
 from abc import (
     ABC,
 )
 from collections.abc import (
     Iterable,
 )
 from typing import (
+    TYPE_CHECKING,
     Final,
     Union,
 )
 
-from crontab import (
-    CronTab,
-)
-
 from .abc import (
     EnrouteDecorator,
 )
 from .kinds import (
     EnrouteDecoratorKind,
 )
 
+if TYPE_CHECKING:
+    from crontab import CronTab as CronTabImpl
+
+    from ...scheduling import (
+        CronTab,
+    )
+
 
 class PeriodicEnrouteDecorator(EnrouteDecorator, ABC):
     """Periodic Enroute class"""
 
-    def __init__(self, crontab: Union[str, CronTab]):
-        if isinstance(crontab, str):
+    def __init__(self, crontab: Union[str, CronTab, CronTabImpl]):
+        from ...scheduling import (
+            CronTab,
+        )
+
+        if not isinstance(crontab, CronTab):
             crontab = CronTab(crontab)
         self.crontab = crontab
 
     def __iter__(self) -> Iterable:
-        yield from (self.crontab.matchers,)
+        yield from (self.crontab,)
 
 
 class PeriodicEventEnrouteDecorator(PeriodicEnrouteDecorator):

packages/core/minos-microservice-networks/minos/networks/scheduling/__init__.py

@@ -1,3 +1,6 @@
+from .crontab import (
+    CronTab,
+)
 from .ports import (
     PeriodicTaskSchedulerPort,
 )

packages/core/minos-microservice-networks/minos/networks/scheduling/crontab.py

@@ -0,0 +1,29 @@
+from __future__ import (
+    annotations,
+)
+
+from typing import (
+    Any,
+    Union,
+)
+
+from crontab import CronTab as CrontTabImpl
+
+
+class CronTab:
+    """TODO"""
+
+    def __init__(self, pattern: Union[str, CrontTabImpl]):
+        if not isinstance(pattern, CrontTabImpl):
+            pattern = CrontTabImpl(pattern)
+        self.impl = pattern
+
+    def __hash__(self):
+        return hash(self.impl.matchers)
+
+    def __eq__(self, other: Any) -> bool:
+        return isinstance(other, type(self)) and self.impl.matchers == other.impl.matchers
+
+    def next(self, *args, **kwargs):
+        """TODO"""
+        self.impl.next(*args, **kwargs)

packages/core/minos-microservice-networks/minos/networks/scheduling/schedulers.py

@@ -21,9 +21,7 @@
     Union,
 )
 
-from crontab import (
-    CronTab,
-)
+from crontab import CronTab as CronTabImpl
 
 from minos.common import (
     Config,
@@ -37,6 +35,9 @@
 from ..requests import (
     ResponseException,
 )
+from .crontab import (
+    CronTab,
+)
 from .requests import (
     ScheduledRequest,
 )
@@ -93,8 +94,8 @@ class PeriodicTask:
 
     _task: Optional[asyncio.Task]
 
-    def __init__(self, crontab: Union[str, CronTab], fn: Callable[[ScheduledRequest], Awaitable[None]]):
-        if isinstance(crontab, str):
+    def __init__(self, crontab: Union[str, CronTab, CronTabImpl], fn: Callable[[ScheduledRequest], Awaitable[None]]):
+        if not isinstance(crontab, CronTab):
             crontab = CronTab(crontab)
 
         self._crontab = crontab

packages/core/minos-microservice-networks/tests/test_networks/test_scheduling/test_schedulers.py

@@ -8,15 +8,12 @@
     patch,
 )
 
-from crontab import (
-    CronTab,
-)
-
 from minos.common import (
     Config,
     current_datetime,
 )
 from minos.networks import (
+    CronTab,
     PeriodicTask,
     PeriodicTaskScheduler,
     ScheduledRequest,
@@ -81,7 +78,7 @@ def setUp(self) -> None:
         self.periodic = PeriodicTask("@daily", self.fn_mock)
 
     def test_crontab(self) -> None:
-        self.assertEqual(CronTab("@daily").matchers, self.periodic.crontab.matchers)
+        self.assertEqual(CronTab("@daily"), self.periodic.crontab)
 
     def test_fn(self) -> None:
         self.assertEqual(self.fn_mock, self.periodic.fn)