Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/scheduler/runner.py: 70%
36 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:48 +0000
1# Code Adapted/Copied From fastapi_utils
2# https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/tasks.py
4import asyncio 1a
5import logging 1a
6from asyncio import ensure_future 1a
7from collections.abc import Callable, Coroutine 1a
8from functools import wraps 1a
9from traceback import format_exception 1a
10from typing import Any 1a
12from starlette.concurrency import run_in_threadpool 1a
14NoArgsNoReturnFuncT = Callable[[], None] 1a
15NoArgsNoReturnAsyncFuncT = Callable[[], Coroutine[Any, Any, None]] 1a
16NoArgsNoReturnDecorator = Callable[[NoArgsNoReturnFuncT | NoArgsNoReturnAsyncFuncT], NoArgsNoReturnAsyncFuncT] 1a
19def repeat_every( 1a
20 *,
21 minutes: float,
22 wait_first: bool = False,
23 logger: logging.Logger | None = None,
24 raise_exceptions: bool = False,
25 max_repetitions: int | None = None,
26) -> NoArgsNoReturnDecorator:
27 """
28 This function returns a decorator that modifies a function so it is periodically re-executed after its first call.
29 The function it decorates should accept no arguments and return nothing. If necessary, this can be accomplished
30 by using `functools.partial` or otherwise wrapping the target function prior to decoration.
32 Parameters
33 ----------
34 seconds: float
35 The number of seconds to wait between repeated calls
36 wait_first: bool (default False)
37 If True, the function will wait for a single period before the first call
38 logger: Optional[logging.Logger] (default None)
39 The logger to use to log any exceptions raised by calls to the decorated function.
40 If not provided, exceptions will not be logged by this function (though they may be handled by the event loop).
41 raise_exceptions: bool (default False)
42 If True, errors raised by the decorated function will be raised to the event loop's exception handler.
43 Note that if an error is raised, the repeated execution will stop.
44 Otherwise, exceptions are just logged and the execution continues to repeat.
45 See https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.set_exception_handler for more info.
46 max_repetitions: Optional[int] (default None)
47 The maximum number of times to call the repeated function. If `None`, the function is repeated forever.
48 """
50 def decorator(func: NoArgsNoReturnAsyncFuncT | NoArgsNoReturnFuncT) -> NoArgsNoReturnAsyncFuncT: 1a
51 """
52 Converts the decorated function into a repeated, periodically-called version of itself.
53 """
54 is_coroutine = asyncio.iscoroutinefunction(func) 1a
56 @wraps(func) 1a
57 async def wrapped() -> None: 1a
58 repetitions = 0 1a
60 async def loop() -> None: 1a
61 nonlocal repetitions
62 if wait_first: 62 ↛ 64line 62 didn't jump to line 64 because the condition on line 62 was always true1a
63 await asyncio.sleep(minutes * 60) 1ab
64 while max_repetitions is None or repetitions < max_repetitions: 64 ↛ exitline 64 didn't return from function 'loop' because the condition on line 64 was always true1bcd
65 try: 1bcd
66 if is_coroutine: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true1bcd
67 await func() # type: ignore
68 else:
69 await run_in_threadpool(func) 1bcd
70 repetitions += 1 1bcd
71 except Exception as exc:
72 if logger is not None:
73 formatted_exception = "".join(format_exception(type(exc), exc, exc.__traceback__))
74 logger.error(formatted_exception)
75 if raise_exceptions:
76 raise exc
77 await asyncio.sleep(minutes * 60) 1bcd
79 ensure_future(loop()) 1a
81 return wrapped 1a
83 return decorator 1a