Coverage for functions \ flipdare \ core \ cron_decorator.py: 95%

19 statements  

« prev     ^ index     » next       coverage.py v7.13.0, created at 2026-05-08 12:22 +1000

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13 

14from collections.abc import Callable 

15from functools import wraps 

16from typing import ParamSpec 

17 

18from flipdare.app_types import CronResult 

19from flipdare.result.app_result import AppResult 

20from flipdare.generated.shared.backend.app_job_type import AppJobType 

21 

22__all__ = ["cron_decorator"] 

23 

24P = ParamSpec("P") 

25 

26 

27def cron_decorator( 

28 job_type: AppJobType, 

29) -> Callable[[Callable[P, "CronResult"]], Callable[P, CronResult]]: 

30 """ 

31 Decorator factory that logs execution results based on error types 

32 and returns a CronResult if successful. 

33 """ 

34 

35 def decorator(func: Callable[P, "CronResult"]) -> Callable[P, CronResult]: 

36 """A decorator to log the start and end of a function call.""" 

37 

38 @wraps(func) 

39 def wrapper(*args: P.args, **kwargs: P.kwargs) -> CronResult: 

40 from flipdare.services import get_app_logger 

41 

42 log_result = func(*args, **kwargs) 

43 if isinstance(log_result, AppResult) and log_result.is_error: 

44 get_app_logger().stat_error( 

45 job_type=job_type, 

46 result=log_result, 

47 ) 

48 return log_result 

49 

50 return wrapper 

51 

52 return decorator