| """ |
| Credential management service for Archon backend (Facade Pattern) |
| Maintains backward compatibility while delegating logic to the credentials package. |
| """ |
|
|
| import os |
| from typing import Any |
|
|
| from ..config.logfire_config import get_logger |
| from .credentials.manager import CredentialItem, CredentialManager |
|
|
| logger = get_logger(__name__) |
|
|
| |
| CredentialItem = CredentialItem |
|
|
|
|
| class CredentialService(CredentialManager): |
| """Facade for CredentialManager to maintain backward compatibility.""" |
|
|
| pass |
|
|
|
|
| |
| credential_service = CredentialService() |
|
|
|
|
| async def get_credential(key: str, default: Any = None) -> Any: |
| """Convenience function to get a credential.""" |
| return await credential_service.get_credential(key, default) |
|
|
|
|
| async def set_credential( |
| key: str, value: str, is_encrypted: bool = False, category: str | None = None, description: str | None = None |
| ) -> bool: |
| """Convenience function to set a credential.""" |
| return await credential_service.set_credential(key, value, is_encrypted, category, description) |
|
|
|
|
| async def initialize_credentials() -> None: |
| """ |
| Initialize the credential service by loading all credentials |
| and setting environment variables (Full Functional Parity). |
| """ |
| await credential_service.load_all_credentials() |
|
|
| |
| infrastructure_credentials = [ |
| "OPENAI_API_KEY", |
| "HOST", |
| "PORT", |
| "MCP_TRANSPORT", |
| "LOGFIRE_ENABLED", |
| "PROJECTS_ENABLED", |
| ] |
|
|
| |
| provider_credentials = [ |
| "GOOGLE_API_KEY", |
| "GEMINI_API_KEY", |
| "LLM_PROVIDER", |
| "LLM_BASE_URL", |
| "EMBEDDING_MODEL", |
| "MODEL_CHOICE", |
| ] |
|
|
| |
| |
|
|
| |
| for key in infrastructure_credentials: |
| try: |
| value = await credential_service.get_credential(key, decrypt=True) |
| if value: |
| os.environ[key] = str(value) |
| logger.debug(f"Mapped infra env: {key}") |
| except Exception as e: |
| logger.warning(f"Failed to set infrastructure env {key}: {e}") |
|
|
| |
| for key in provider_credentials: |
| try: |
| value = await credential_service.get_credential(key, decrypt=True) |
| if value: |
| env_key = key.upper() |
| os.environ[env_key] = str(value) |
| logger.debug(f"Mapped provider env: {env_key}") |
| except Exception: |
| pass |
|
|
| logger.info("✅ Credentials initialized and physical environment mapped") |
|
|