app/services/ping.py

from __future__ import annotations

import asyncio
from typing import Optional

import httpx
import structlog

from app.core.config import settings

logger = structlog.get_logger(__name__)


async def start_self_ping() -> Optional[asyncio.Task]:
    if not settings.SELF_PING_ENABLED or not settings.SELF_PING_URL:
        logger.info("self_ping_disabled")
        return None
    task = asyncio.create_task(_ping_loop())
    logger.info("self_ping_started", url=settings.SELF_PING_URL, interval=settings.SELF_PING_INTERVAL_SECONDS)
    return task


async def _ping_loop() -> None:
    logger.info(
        "self_ping_started",
        url=settings.SELF_PING_URL,
        interval=settings.SELF_PING_INTERVAL_SECONDS,
    )
    async with httpx.AsyncClient(timeout=10.0) as client:
        while True:
            await asyncio.sleep(settings.SELF_PING_INTERVAL_SECONDS)
            await _execute_ping(client)


async def _execute_ping(client: httpx.AsyncClient) -> None:
    headers: dict[str, str] = {}
    if settings.HF_TOKEN:
        headers["Authorization"] = f"Bearer {settings.HF_TOKEN}"

    try:
        response = await client.get(settings.SELF_PING_URL, headers=headers)
        logger.info("self_ping_success", status_code=response.status_code)
    except httpx.RequestError as exc:
        logger.warning("self_ping_failed", error=str(exc))