Coverage for functions \ flipdare \ app_cron.py: 100%
0 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#
13# pyright: reportCallIssue=false
14# NOTE: triggers should be as lightweight as possible, just creating jobs for processing
15"""
16!! Optimal Maintenance Windows (US Time Zones)
17NOTE: Best Time: 1:00 AM - 4:00 AM Eastern Time.
18NOTE: Alternative Time: - 2:00 AM - 3:00 AM Monday mornings are often considered optimal for low traffic.
20NOTE: Weekend Option: Sunday nights, which align with the lowest activity, allow for longer, more extensive maintenance tasks
22"""
24from zoneinfo import ZoneInfo
26from firebase_functions import scheduler_fn as sched # pragma: no cover
28from flipdare.generated.shared.backend.app_job_type import AppJobType
29from flipdare.job.app_job_schedule import AppJobSchedule # pragma: no cover
30from flipdare.job.cron_validator import CronValidator # pragma: no cover
31from flipdare.services import get_job_admin # pragma: no cover
33__all__ = [ # pragma: no cover
34 "fn_two_minutes",
35 "fn_five_minutes",
36 "fn_fifteen_minutes",
37 "fn_one_hour",
38 "fn_four_hours",
39 "fn_six_hours",
40 "fn_twelve_hours",
41 "fn_one_day",
42 "fn_one_week",
43 "fn_three_days",
44]
46# note, need this to stop pylance intermittently warning about "on_schedule" ..
49@sched.on_schedule(
50 schedule="every 2 minutes synchronized",
51 timezone=ZoneInfo("America/Los_Angeles"),
52)
53def fn_two_minutes(event: sched.ScheduledEvent) -> None: # pragma: no cover
54 """Every 2 minutes."""
55 _run(AppJobType.SCHED_TWO_MINS, event, AppJobSchedule.TWO_MINUTES)
58@sched.on_schedule(
59 schedule="every 5 minutes synchronized",
60 timezone=ZoneInfo("America/Los_Angeles"),
61)
62def fn_five_minutes(event: sched.ScheduledEvent) -> None: # pragma: no cover
63 """Every 5 minutes."""
64 _run(AppJobType.SCHED_FIVE_MINS, event, AppJobSchedule.FIVE_MINUTES)
67@sched.on_schedule(
68 schedule="every 15 minutes synchronized",
69 timezone=ZoneInfo("America/Los_Angeles"),
70)
71def fn_fifteen_minutes(event: sched.ScheduledEvent) -> None: # pragma: no cover
72 """Every 15 minutes."""
73 _run(AppJobType.SCHED_FIFTEEN_MINS, event, AppJobSchedule.FIFTEEN_MINUTES)
76@sched.on_schedule(
77 schedule="every hour synchronized",
78 timezone=ZoneInfo("America/Los_Angeles"),
79)
80def fn_one_hour(event: sched.ScheduledEvent) -> None: # pragma: no cover
81 """Every hour."""
82 _run(AppJobType.SCHED_ONE_HOUR, event, AppJobSchedule.ONE_HOUR)
85@sched.on_schedule(
86 schedule="every 2 hours synchronized",
87 timezone=ZoneInfo("America/Los_Angeles"),
88)
89def fn_two_hours(event: sched.ScheduledEvent) -> None: # pragma: no cover
90 """Every 2 hours."""
91 _run(AppJobType.SCHED_TWO_HOURS, event, AppJobSchedule.TWO_HOURS)
94@sched.on_schedule(
95 schedule="every 4 hours synchronized",
96 timezone=ZoneInfo("America/Los_Angeles"),
97)
98def fn_four_hours(event: sched.ScheduledEvent) -> None: # pragma: no cover
99 """Every 4 hours."""
100 _run(AppJobType.SCHED_FOUR_HOURS, event, AppJobSchedule.FOUR_HOURS)
103@sched.on_schedule(
104 schedule="every 6 hours synchronized",
105 timezone=ZoneInfo("America/Los_Angeles"),
106)
107def fn_six_hours(event: sched.ScheduledEvent) -> None: # pragma: no cover
108 """Every 6 hours."""
109 _run(AppJobType.SCHED_SIX_HOURS, event, AppJobSchedule.SIX_HOURS)
112@sched.on_schedule(
113 schedule="every 12 hours synchronized",
114 timezone=ZoneInfo("America/Los_Angeles"),
115)
116def fn_twelve_hours(event: sched.ScheduledEvent) -> None: # pragma: no cover
117 """Every 12 hours."""
118 _run(AppJobType.SCHED_TWELVE_HOURS, event, AppJobSchedule.TWELVE_HOURS)
121@sched.on_schedule(schedule="every day 01:00", timezone=ZoneInfo("America/Los_Angeles"))
122def fn_one_day(event: sched.ScheduledEvent) -> None: # pragma: no cover
123 """Every day."""
124 _run(AppJobType.SCHED_ONE_DAY, event, AppJobSchedule.ONE_DAY)
127# Every 3 days at 1:30 AM (adjust as needed for optimal timing)
128@sched.on_schedule(schedule="30 1 */3 * *", timezone=ZoneInfo("America/Los_Angeles"))
129def fn_three_days(event: sched.ScheduledEvent) -> None: # pragma: no cover
130 """Every three days."""
131 _run(AppJobType.SCHED_THREE_DAYS, event, AppJobSchedule.THREE_DAYS)
134@sched.on_schedule(schedule="every sunday 03:00", timezone=ZoneInfo("America/Los_Angeles"))
135def fn_one_week(event: sched.ScheduledEvent) -> None: # pragma: no cover
136 """Every 7 days."""
137 _run(AppJobType.SCHED_ONE_WEEK, event, AppJobSchedule.ONE_WEEK)
140def _run(
141 job_type: AppJobType,
142 event: sched.ScheduledEvent,
143 interval: AppJobSchedule,
144) -> None: # pragma: no cover
145 validator = CronValidator(job_type, event)
146 get_job_admin().cron_job(job_type, validator, interval)