admin.py 2.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495
  1. """
  2. Admin service for handling administrative operations.
  3. """
  4. import logging
  5. from typing import List
  6. from fastapi import Depends, Request
  7. from web.core.config import get_settings, Settings
  8. logger = logging.getLogger(__name__)
  9. class AdminService:
  10. """Service for managing administrative operations."""
  11. def __init__(self, remote_reload_token: str, versions_fetcher=None):
  12. """
  13. Initialize the admin service.
  14. Args:
  15. remote_reload_token: Remote reload authentication token
  16. versions_fetcher: VersionsFetcher instance for managing remotes
  17. """
  18. self.remote_reload_token = remote_reload_token
  19. self.versions_fetcher = versions_fetcher
  20. async def verify_remote_reload_token(self, token: str) -> bool:
  21. """
  22. Verify that the provided token matches the expected remote reload token.
  23. Args:
  24. token: The token to verify
  25. Returns:
  26. True if token is valid, False otherwise
  27. """
  28. if not token:
  29. return False
  30. return token == self.remote_reload_token
  31. async def refresh_remotes(self) -> List[str]:
  32. """
  33. Trigger a refresh of remote metadata.
  34. Returns:
  35. List of remote names that were refreshed
  36. Raises:
  37. Exception: If refresh operation fails
  38. """
  39. logger.info("Triggering remote metadata refresh")
  40. # Reload remotes.json
  41. self.versions_fetcher.reload_remotes_json()
  42. # Get list of remotes that are now available
  43. remotes_info = self.versions_fetcher.get_all_remotes_info()
  44. remotes_refreshed = [remote.name for remote in remotes_info]
  45. logger.info(
  46. f"Successfully refreshed {len(remotes_refreshed)} remotes: "
  47. f"{remotes_refreshed}"
  48. )
  49. return remotes_refreshed
  50. def get_admin_service(
  51. request: Request,
  52. settings: Settings = Depends(get_settings)
  53. ) -> AdminService:
  54. """
  55. Get AdminService instance with dependencies from app state.
  56. Args:
  57. request: FastAPI Request object
  58. settings: Application settings
  59. Returns:
  60. AdminService instance initialized with app state dependencies
  61. Raises:
  62. RuntimeError: If remote reload token is not configured
  63. """
  64. remote_reload_token = settings.remote_reload_token
  65. if remote_reload_token is None:
  66. raise RuntimeError("Remote reload token not configured on server")
  67. return AdminService(
  68. remote_reload_token=remote_reload_token,
  69. versions_fetcher=request.app.state.versions_fetcher
  70. )