| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495 |
- """
- Admin service for handling administrative operations.
- """
- import logging
- from typing import List
- from fastapi import Depends, Request
- from web.core.config import get_settings, Settings
- logger = logging.getLogger(__name__)
- class AdminService:
- """Service for managing administrative operations."""
- def __init__(self, remote_reload_token: str, versions_fetcher=None):
- """
- Initialize the admin service.
- Args:
- remote_reload_token: Remote reload authentication token
- versions_fetcher: VersionsFetcher instance for managing remotes
- """
- self.remote_reload_token = remote_reload_token
- self.versions_fetcher = versions_fetcher
- async def verify_remote_reload_token(self, token: str) -> bool:
- """
- Verify that the provided token matches the expected remote reload token.
- Args:
- token: The token to verify
- Returns:
- True if token is valid, False otherwise
- """
- if not token:
- return False
- return token == self.remote_reload_token
- async def refresh_remotes(self) -> List[str]:
- """
- Trigger a refresh of remote metadata.
- Returns:
- List of remote names that were refreshed
- Raises:
- Exception: If refresh operation fails
- """
- logger.info("Triggering remote metadata refresh")
- # Reload remotes.json
- self.versions_fetcher.reload_remotes_json()
- # Get list of remotes that are now available
- remotes_info = self.versions_fetcher.get_all_remotes_info()
- remotes_refreshed = [remote.name for remote in remotes_info]
- logger.info(
- f"Successfully refreshed {len(remotes_refreshed)} remotes: "
- f"{remotes_refreshed}"
- )
- return remotes_refreshed
- def get_admin_service(
- request: Request,
- settings: Settings = Depends(get_settings)
- ) -> AdminService:
- """
- Get AdminService instance with dependencies from app state.
- Args:
- request: FastAPI Request object
- settings: Application settings
- Returns:
- AdminService instance initialized with app state dependencies
- Raises:
- RuntimeError: If remote reload token is not configured
- """
- remote_reload_token = settings.remote_reload_token
- if remote_reload_token is None:
- raise RuntimeError("Remote reload token not configured on server")
- return AdminService(
- remote_reload_token=remote_reload_token,
- versions_fetcher=request.app.state.versions_fetcher
- )
|