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
« 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#
14from collections.abc import Callable
15from functools import wraps
16from typing import ParamSpec
18from flipdare.app_types import CronResult
19from flipdare.result.app_result import AppResult
20from flipdare.generated.shared.backend.app_job_type import AppJobType
22__all__ = ["cron_decorator"]
24P = ParamSpec("P")
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 """
35 def decorator(func: Callable[P, "CronResult"]) -> Callable[P, CronResult]:
36 """A decorator to log the start and end of a function call."""
38 @wraps(func)
39 def wrapper(*args: P.args, **kwargs: P.kwargs) -> CronResult:
40 from flipdare.services import get_app_logger
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
50 return wrapper
52 return decorator