# [DEF:backend.src.api.routes.environments:Module] # # @SEMANTICS: api, environments, superset, databases # @PURPOSE: API endpoints for listing environments and their databases. # @LAYER: API # @RELATION: DEPENDS_ON -> backend.src.dependencies # @RELATION: DEPENDS_ON -> backend.src.core.superset_client # # @INVARIANT: Environment IDs must exist in the configuration. # [SECTION: IMPORTS] from fastapi import APIRouter, Depends, HTTPException from typing import List, Dict, Optional from backend.src.dependencies import get_config_manager from backend.src.core.superset_client import SupersetClient from superset_tool.models import SupersetConfig from pydantic import BaseModel # [/SECTION] router = APIRouter(prefix="/api/environments", tags=["environments"]) # [DEF:EnvironmentResponse:DataClass] class EnvironmentResponse(BaseModel): id: str name: str url: str # [/DEF:EnvironmentResponse] # [DEF:DatabaseResponse:DataClass] class DatabaseResponse(BaseModel): uuid: str database_name: str engine: Optional[str] # [/DEF:DatabaseResponse] # [DEF:get_environments:Function] # @PURPOSE: List all configured environments. # @RETURN: List[EnvironmentResponse] @router.get("", response_model=List[EnvironmentResponse]) async def get_environments(config_manager=Depends(get_config_manager)): envs = config_manager.get_environments() return [EnvironmentResponse(id=e.id, name=e.name, url=e.url) for e in envs] # [/DEF:get_environments] # [DEF:get_environment_databases:Function] # @PURPOSE: Fetch the list of databases from a specific environment. # @PARAM: id (str) - The environment ID. # @RETURN: List[Dict] - List of databases. @router.get("/{id}/databases") async def get_environment_databases(id: str, config_manager=Depends(get_config_manager)): envs = config_manager.get_environments() env = next((e for e in envs if e.id == id), None) if not env: raise HTTPException(status_code=404, detail="Environment not found") try: # Initialize SupersetClient from environment config # Note: We need to map Environment model to SupersetConfig superset_config = SupersetConfig( env=env.name, base_url=env.url, auth={ "provider": "db", # Defaulting to db provider "username": env.username, "password": env.password, "refresh": "false" } ) client = SupersetClient(superset_config) return client.get_databases_summary() except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to fetch databases: {str(e)}") # [/DEF:get_environment_databases] # [/DEF:backend.src.api.routes.environments]