🚀cron jobs with repeat_at (#3)

* 🚀cron jobs with `repeat_at`

* 🧪Test Cases updated

* added `cronitier` to requirements

* 🧪Test Cases added

Author: priyanshu-panwar

README.md

@@ -23,16 +23,26 @@
 
 This package includes a number of utilities to help reduce boilerplate and reuse common functionality across projects:
 
-* **Repeated Tasks**: Easily trigger periodic tasks on server startup.
+* **Repeated Tasks**: Easily trigger periodic tasks on server startup using **repeat_every**.
 ```
-from fastapi_utilities.repeat import repeat_every
+from fastapi_utilities import repeat_every
 
 @router.on_event('startup')
 @repeat_every(seconds=3)
 async def print_hello():
     print("hello")
 ```
 
+* **Cron Jobs**: Easily trigger cron jobs on server startup using **repeat_at** by providing a cron expression.
+```
+from fastapi_utilities import repeat_at
+
+@router.on_event("startup")
+@repeat_at(cron="*/2 * * * *") #every 2nd minute
+async def hey():
+    print("hey")
+```
+
 ---
 
 ## Requirements

fastapi_utilities/__init__.py

@@ -0,0 +1,2 @@
+from .repeat.repeat_every import repeat_every
+from .repeat.repeat_at import repeat_at

fastapi_utilities/repeat/__init__.py

@@ -0,0 +1,73 @@
+import asyncio
+import logging
+
+from datetime import datetime
+from functools import wraps
+from asyncio import ensure_future
+from starlette.concurrency import run_in_threadpool
+from croniter import croniter
+
+
+def get_delta(cron):
+    """
+    This function returns the time delta between now and the next cron execution time.
+    """
+    now = datetime.now()
+    cron = croniter(cron, now)
+    return (cron.get_next(datetime) - now).total_seconds()
+
+
+def repeat_at(
+    *,
+    cron: str,
+    logger: logging.Logger = None,
+    raise_exceptions: bool = False,
+    max_repetitions: int = None,
+):
+    """
+    This function returns a decorator that makes a function execute periodically as per the cron expression provided.
+
+    :: Params ::
+    ------------
+    cron: str
+        Cron-style string for periodic execution, eg. '0 0 * * *' every midnight
+    logger: logging.Logger (default None)
+        Logger object to log exceptions
+    raise_exceptions: bool (default False)
+        Whether to raise exceptions or log them
+    max_repetitions: int (default None)
+        Maximum number of times to repeat the function. If None, repeat indefinitely.
+
+    """
+
+    def decorator(func):
+        is_coroutine = asyncio.iscoroutinefunction(func)
+
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            repititions = 0
+            if not croniter.is_valid(cron):
+                raise ValueError("Invalid cron expression")
+
+            async def loop(*args, **kwargs):
+                nonlocal repititions
+                while max_repetitions is None or repititions < max_repetitions:
+                    try:
+                        sleepTime = get_delta(cron)
+                        await asyncio.sleep(sleepTime)
+                        if is_coroutine:
+                            await func(*args, **kwargs)
+                        else:
+                            await run_in_threadpool(func, *args, **kwargs)
+                    except Exception as e:
+                        if logger is not None:
+                            logger.exception(e)
+                        if raise_exceptions:
+                            raise e
+                    repititions += 1
+
+            ensure_future(loop(*args, **kwargs))
+
+        return wrapper
+
+    return decorator

fastapi_utilities/repeat/repeat_at.py

@@ -32,6 +32,7 @@ python = "^3.7"
 fastapi = "*"
 pydantic = "*"
 sqlalchemy = "*"
+croniter = "^1.0.13"
 
 [tool.poetry.group.dev.dependencies]
 # Testing

fastapi_utilities/repeat.py renamed to fastapi_utilities/repeat/repeat_every.py

@@ -226,6 +226,9 @@ coverage[toml]==7.2.7 ; python_version >= "3.7" and python_version < "4.0" \
     --hash=sha256:f6951407391b639504e3b3be51b7ba5f3528adbf1a8ac3302b687ecababf929e \
     --hash=sha256:f75f7168ab25dd93110c8a8117a22450c19976afbc44234cbf71481094c1b850 \
     --hash=sha256:fdec9e8cbf13a5bf63290fc6013d216a4c7232efb51548594ca3631a7f13c3a3
+croniter==1.4.1 ; python_version >= "3.7" and python_version < "4.0" \
+    --hash=sha256:1a6df60eacec3b7a0aa52a8f2ef251ae3dd2a7c7c8b9874e73e791636d55a361 \
+    --hash=sha256:9595da48af37ea06ec3a9f899738f1b2c1c13da3c38cea606ef7cd03ea421128
 exceptiongroup==1.1.3 ; python_version >= "3.7" and python_version < "3.11" \
     --hash=sha256:097acd85d473d75af5bb98e41b61ff7fe35efe6675e4f9370ec6ec5126d160e9 \
     --hash=sha256:343280667a4585d195ca1cf9cef84a4e178c4b6cf2274caef9859782b567d5e3
@@ -430,9 +433,15 @@ pytest-cov==4.1.0 ; python_version >= "3.7" and python_version < "4.0" \
 pytest==7.4.2 ; python_version >= "3.7" and python_version < "4.0" \
     --hash=sha256:1d881c6124e08ff0a1bb75ba3ec0bfd8b5354a01c194ddd5a0a870a48d99b002 \
     --hash=sha256:a766259cfab564a2ad52cb1aae1b881a75c3eb7e34ca3779697c23ed47c47069
+python-dateutil==2.8.2 ; python_version >= "3.7" and python_version < "4.0" \
+    --hash=sha256:0123cacc1627ae19ddf3c27a5de5bd67ee4586fbdd6440d9748f8abb483d3e86 \
+    --hash=sha256:961d03dc3453ebbc59dbdea9e4e11c5651520a876d0f4db161e8674aae935da9
 requests==2.31.0 ; python_version >= "3.7" and python_version < "4.0" \
     --hash=sha256:58cd2187c01e70e6e26505bca751777aa9f2ee0b7f4300988b709f44e013003f \
     --hash=sha256:942c5a758f98d790eaed1a29cb6eefc7ffb0d1cf7af05c3d2791656dbd6ad1e1
+six==1.16.0 ; python_version >= "3.7" and python_version < "4.0" \
+    --hash=sha256:1e61c37477a1626458e36f7b1d82aa5c9b094fa4802892072e49de9c60c4c926 \
+    --hash=sha256:8abb2f1d86890a2dfb989f9a77cfcfd3e47c2a354b01111771326f8aa26e0254
 sniffio==1.3.0 ; python_version >= "3.7" and python_version < "4.0" \
     --hash=sha256:e60305c5e5d314f5389259b7f22aaa33d8f7dee49763119234af3755c55b9101 \
     --hash=sha256:eecefdce1e5bbfb7ad2eeaabf7c1eeb404d7757c379bd1f7e5cce9d8bf425384

poetry.lock

@@ -0,0 +1,117 @@
+import logging
+import asyncio
+import pytest
+from _pytest.capture import CaptureFixture
+from _pytest.logging import LogCaptureFixture
+from fastapi_utilities import repeat_at
+
+
+@pytest.mark.asyncio
+async def test_repeat_at(capsys: CaptureFixture[str]):
+    """
+    Simple Test Case for repeat_at
+    """
+
+    @repeat_at(cron="* * * * *", max_repetitions=3)
+    async def print_hello():
+        print("Hello")
+
+    print_hello()
+    await asyncio.sleep(1)
+    out, err = capsys.readouterr()
+    assert err == ""
+    assert out == ""
+
+
+@pytest.mark.asyncio
+async def test_repeat_at_def(capsys: CaptureFixture[str]):
+    """
+    Simple Test Case for repeat_at
+    """
+
+    @repeat_at(cron="* * * * *", max_repetitions=3)
+    def print_hello():
+        print("Hello")
+
+    print_hello()
+    await asyncio.sleep(1)
+    out, err = capsys.readouterr()
+    assert err == ""
+    assert out == ""
+
+
+@pytest.mark.asyncio
+async def test_repeat_at_with_logger(caplog: LogCaptureFixture):
+    """
+    Test Case for repeat_at with logger
+    """
+
+    @repeat_at(cron="* * * * *", logger=logging.getLogger("test"), max_repetitions=3)
+    async def print_hello():
+        raise Exception("Hello")
+
+    print_hello()
+    await asyncio.sleep(60)
+
+    captured_logs = caplog.records
+
+    assert len(captured_logs) > 0
+
+
+from asyncio import AbstractEventLoop
+from typing import Any, Dict
+
+
+def ignore_exception(_loop: AbstractEventLoop, _context: Dict[str, Any]) -> None:
+    pass
+
+
+@pytest.fixture(autouse=True)
+def setup_event_loop(event_loop: AbstractEventLoop) -> None:
+    event_loop.set_exception_handler(ignore_exception)
+
+
+@pytest.mark.asyncio
+async def test_repeat_at_exception(capsys: CaptureFixture[str]) -> None:
+    """
+    Test Case for repeat_at with an invalid cron expression
+    """
+
+    logger = logging.getLogger(__name__)
+
+    @repeat_at(
+        cron="* * * * *", max_repetitions=None, raise_exceptions=True, logger=logger
+    )
+    def raise_exc():
+        raise ValueError("repeat")
+
+    try:
+        raise_exc()
+        await asyncio.sleep(60)
+    except ValueError:
+        out, err = capsys.readouterr()
+        assert out == ""
+        assert err == ""
+
+
+@pytest.mark.asyncio
+async def test_repeat_at_invalid_cron(capsys: CaptureFixture[str]) -> None:
+    """
+    Test Case for repeat_at with an invalid cron expression
+    """
+
+    logger = logging.getLogger(__name__)
+
+    @repeat_at(
+        cron="invalid", max_repetitions=None, raise_exceptions=True, logger=logger
+    )
+    def raise_exc():
+        raise ValueError("repeat")
+
+    try:
+        await raise_exc()
+        await asyncio.sleep(60)
+    except ValueError:
+        out, err = capsys.readouterr()
+        assert out == ""
+        assert err == ""

pyproject.toml

@@ -4,7 +4,7 @@
 
 from _pytest.capture import CaptureFixture
 from _pytest.logging import LogCaptureFixture
-from fastapi_utilities.repeat import repeat_every
+from fastapi_utilities import repeat_every
 
 
 @pytest.mark.asyncio