openswarm/backend/apps/skills/RegistryRefreshLoop/RegistryRefreshLoop.py

import asyncio
import time
from typing import Optional

from swarm_debug import debug
from typeguard import typechecked
from pydantic import BaseModel, Field, InstanceOf

from backend.apps.skills.RegistryRefreshLoop.fetch_all_registry_skills.fetch_all_registry_skills import fetch_all_registry_skills


class RegistryRefreshLoop(BaseModel):
    refresh_interval_s: int
    num_concurrent_fetches: int
    github_base_url: str
    github_repo: str
    github_branch: str
    manifest_extension: str
    cache: dict[str, dict] = Field(default_factory=dict)
    updated_at: float = 0
    task: Optional[InstanceOf[asyncio.Task]] = None

    @typechecked
    async def start(self) -> None:
        self._task = asyncio.create_task(self.p_run_loop())

    @typechecked
    async def stop(self) -> None:
        if self._task:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass

    @typechecked
    async def p_run_loop(self) -> None:
        while True:
            try:
                self.cache = await fetch_all_registry_skills(
                    num_concurrent_fetches=self.num_concurrent_fetches,
                    github_base_url=self.github_base_url,
                    github_repo=self.github_repo,
                    github_branch=self.github_branch,
                    manifest_extension=self.manifest_extension,
                )
                self.updated_at = time.time()
            except Exception as e:
                debug(f"[RegistryRefreshLoop] Skill registry refresh error: {e}")
            await asyncio.sleep(self.refresh_interval_s)