Spaces:
Sleeping
Sleeping
File size: 1,357 Bytes
936f0bf 697be7d 936f0bf 697be7d 936f0bf 697be7d 936f0bf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 | 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))
|