Исходный код soniks_client.jobs

"""Регистрация постоянных заданий планировщика."""

from datetime import UTC, datetime

from apscheduler.schedulers.base import BaseScheduler

from core.configs import settings
from soniks_client.jobs.resending import resending_jobs
from soniks_client.jobs.sync import sync_observation_jobs
from soniks_client.sat_data import resync_sat_data
from soniks_client.scheduler import SYNC_JOB_ID




def register_observation_jobs(scheduler: BaseScheduler) -> None:
    """Зарегистрировать периодическую сверку расписания с порталом."""

    scheduler.add_job(
        func=sync_observation_jobs,
        trigger="interval",
        minutes=settings.scheduler.JOB_REQUEST_INTERVAL_IN_MINUTES,
        next_run_time=datetime.now(UTC),
        id=SYNC_JOB_ID,
        replace_existing=True,
        args=[scheduler],
    )





def register_resending_jobs(scheduler: BaseScheduler) -> None:
    """Зарегистрировать периодическую повторную отправку из ``incomplete``."""

    scheduler.add_job(
        func=resending_jobs,
        trigger="interval",
        minutes=settings.scheduler.RESENDING_INTERVAL_IN_MINUTES,
        next_run_time=datetime.now(UTC),
        id="register_resending_jobs",
        replace_existing=True,
        args=[scheduler],
    )





def register_sat_data_jobs(scheduler: BaseScheduler) -> None:
    """Зарегистрировать периодическое обновление определений спутников.

    Без ``next_run_time``: на старте bundle уже синхронизирован из ``main()``.
    """

    scheduler.add_job(
        func=resync_sat_data,
        trigger="interval",
        hours=settings.scheduler.SAT_DATA_SYNC_INTERVAL_IN_HOURS,
        id="register_sat_data_jobs",
        replace_existing=True,
    )