config.py 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134
  1. """
  2. Application configuration and settings.
  3. """
  4. import os
  5. import logging
  6. from pathlib import Path
  7. from functools import lru_cache
  8. from typing import Optional
  9. logger = logging.getLogger(__name__)
  10. class Settings:
  11. """Application settings."""
  12. def __init__(self):
  13. # Application
  14. self.app_name: str = "CustomBuild API"
  15. self.app_version: str = "1.0.0"
  16. self.debug: bool = False
  17. # Paths
  18. self.base_dir: str = os.getenv(
  19. "CBS_BASEDIR",
  20. default=str(Path(__file__).parent.parent.parent.parent / "base")
  21. )
  22. # Redis
  23. self.redis_host: str = os.getenv(
  24. 'CBS_REDIS_HOST',
  25. default='localhost'
  26. )
  27. self.redis_port: str = os.getenv(
  28. 'CBS_REDIS_PORT',
  29. default='6379'
  30. )
  31. # Logging
  32. self.log_level: str = os.getenv('CBS_LOG_LEVEL', default='INFO')
  33. # ArduPilot Git Repository
  34. self.ap_git_url: str = "https://github.com/ardupilot/ardupilot.git"
  35. @property
  36. def source_dir(self) -> str:
  37. """ArduPilot source directory."""
  38. return os.path.join(self.base_dir, 'ardupilot')
  39. @property
  40. def artifacts_dir(self) -> str:
  41. """Build artifacts directory."""
  42. return os.path.join(self.base_dir, 'artifacts')
  43. @property
  44. def outdir_parent(self) -> str:
  45. """Build output directory (same as artifacts_dir)."""
  46. return self.artifacts_dir
  47. @property
  48. def workdir_parent(self) -> str:
  49. """Work directory parent."""
  50. return os.path.join(self.base_dir, 'workdir')
  51. @property
  52. def remotes_json_path(self) -> str:
  53. """Path to remotes.json configuration."""
  54. return os.path.join(self.base_dir, 'configs', 'remotes.json')
  55. @property
  56. def rate_limiter_storage_uri(self) -> str:
  57. """
  58. Storage URI for the rate limiter.
  59. Uses CBS_RATE_LIMITER_STORAGE_URI env var if set and non-empty,
  60. otherwise falls back to redis backend. When using the redis backend,
  61. db index 1 is used to avoid conflicts with any other components that
  62. use db index 0.
  63. """
  64. uri = os.getenv('CBS_RATE_LIMITER_STORAGE_URI', '').strip()
  65. if uri:
  66. return uri
  67. return f"redis://{self.redis_host}:{self.redis_port}/1"
  68. @property
  69. def enable_inbuilt_builder(self) -> bool:
  70. """Whether to enable the inbuilt builder."""
  71. return os.getenv('CBS_ENABLE_INBUILT_BUILDER', '1') == '1'
  72. @property
  73. def manifest_cache_dir(self) -> str:
  74. """Directory for cached firmware manifest files."""
  75. return os.path.dirname(self.remotes_json_path)
  76. @property
  77. def ap_firmware_manifest_url(self) -> str:
  78. """URL for the ArduPilot firmware manifest."""
  79. return os.getenv(
  80. 'CBS_AP_FIRMWARE_MANIFEST_URL',
  81. 'https://firmware.ardupilot.org/manifest.json.xz',
  82. )
  83. @property
  84. def admin_token(self) -> Optional[str]:
  85. """
  86. Get admin API token from file or environment variable.
  87. Tries to read token from file first, falls back to environment variable.
  88. Returns:
  89. The authorization token if found, None otherwise
  90. """
  91. token_file_path = os.path.join(self.base_dir, 'secrets', 'admin_token')
  92. try:
  93. with open(token_file_path, 'r') as file:
  94. token = file.read().strip()
  95. return token
  96. except (FileNotFoundError, PermissionError):
  97. env_token = os.getenv('CBS_ADMIN_TOKEN', '')
  98. return env_token if env_token != '' else None
  99. except Exception as e:
  100. logger.error(
  101. f"Unexpected error reading token file at {token_file_path}: {e}. "
  102. "Checking environment for token."
  103. )
  104. env_token = os.getenv('CBS_ADMIN_TOKEN', None)
  105. return env_token if env_token != '' else None
  106. @lru_cache()
  107. def get_settings() -> Settings:
  108. """Get cached settings instance."""
  109. return Settings()