admin.py 2.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293
  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, admin_token: str, versions_manager=None):
  12. """
  13. Initialize the admin service.
  14. Args:
  15. admin_token: Admin API authentication token
  16. versions_manager: VersionsManager instance for managing remotes
  17. """
  18. self.admin_token = admin_token
  19. self.versions_manager = versions_manager
  20. async def verify_admin_token(self, token: str) -> bool:
  21. """
  22. Verify that the provided token matches the expected admin 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.admin_token
  31. async def refresh_versions(self) -> List[str]:
  32. """
  33. Refresh all version providers and sync git remotes.
  34. Returns:
  35. List of git remote names synced after the refresh
  36. Raises:
  37. Exception: If refresh operation fails
  38. """
  39. logger.info("Triggering version provider refresh")
  40. self.versions_manager.refresh_all()
  41. remotes_info = self.versions_manager.get_all_remotes_info()
  42. remotes_synced = [remote.name for remote in remotes_info]
  43. logger.info(
  44. f"Successfully refreshed version providers; "
  45. f"synced {len(remotes_synced)} remotes: {remotes_synced}"
  46. )
  47. return remotes_synced
  48. def get_admin_service(
  49. request: Request,
  50. settings: Settings = Depends(get_settings)
  51. ) -> AdminService:
  52. """
  53. Get AdminService instance with dependencies from app state.
  54. Args:
  55. request: FastAPI Request object
  56. settings: Application settings
  57. Returns:
  58. AdminService instance initialized with app state dependencies
  59. Raises:
  60. RuntimeError: If admin token is not configured
  61. """
  62. admin_token = settings.admin_token
  63. if admin_token is None:
  64. raise RuntimeError("Admin token not configured on server")
  65. return AdminService(
  66. admin_token=admin_token,
  67. versions_manager=request.app.state.versions_manager
  68. )