84 lines
3.1 KiB
Python
84 lines
3.1 KiB
Python
# [DEF:backend.src.core.superset_client:Module]
|
|
#
|
|
# @SEMANTICS: superset, api, client, database, metadata
|
|
# @PURPOSE: Extends the base SupersetClient with database-specific metadata fetching.
|
|
# @LAYER: Core
|
|
# @RELATION: INHERITS_FROM -> superset_tool.client.SupersetClient
|
|
#
|
|
# @INVARIANT: All database metadata requests must include UUID and name.
|
|
|
|
# [SECTION: IMPORTS]
|
|
from typing import List, Dict, Optional, Tuple
|
|
from superset_tool.client import SupersetClient as BaseSupersetClient
|
|
from superset_tool.models import SupersetConfig
|
|
# [/SECTION]
|
|
|
|
# [DEF:SupersetClient:Class]
|
|
# @PURPOSE: Extended SupersetClient for migration-specific operations.
|
|
class SupersetClient(BaseSupersetClient):
|
|
|
|
# [DEF:SupersetClient.get_databases_summary:Function]
|
|
# @PURPOSE: Fetch a summary of databases including uuid, name, and engine.
|
|
# @POST: Returns a list of database dictionaries with 'engine' field.
|
|
# @RETURN: List[Dict] - Summary of databases.
|
|
def get_databases_summary(self) -> List[Dict]:
|
|
"""
|
|
Fetch a summary of databases including uuid, name, and engine.
|
|
"""
|
|
query = {
|
|
"columns": ["uuid", "database_name", "backend"]
|
|
}
|
|
_, databases = self.get_databases(query=query)
|
|
|
|
# Map 'backend' to 'engine' for consistency with contracts
|
|
for db in databases:
|
|
db['engine'] = db.pop('backend', None)
|
|
|
|
return databases
|
|
# [/DEF:SupersetClient.get_databases_summary]
|
|
|
|
# [DEF:SupersetClient.get_database_by_uuid:Function]
|
|
# @PURPOSE: Find a database by its UUID.
|
|
# @PARAM: db_uuid (str) - The UUID of the database.
|
|
# @RETURN: Optional[Dict] - Database info if found, else None.
|
|
def get_database_by_uuid(self, db_uuid: str) -> Optional[Dict]:
|
|
"""
|
|
Find a database by its UUID.
|
|
"""
|
|
query = {
|
|
"filters": [{"col": "uuid", "op": "eq", "value": db_uuid}]
|
|
}
|
|
_, databases = self.get_databases(query=query)
|
|
return databases[0] if databases else None
|
|
# [/DEF:SupersetClient.get_database_by_uuid]
|
|
|
|
# [DEF:SupersetClient.get_dashboards_summary:Function]
|
|
# @PURPOSE: Fetches dashboard metadata optimized for the grid.
|
|
# @POST: Returns a list of dashboard dictionaries.
|
|
# @RETURN: List[Dict]
|
|
def get_dashboards_summary(self) -> List[Dict]:
|
|
"""
|
|
Fetches dashboard metadata optimized for the grid.
|
|
Returns a list of dictionaries mapped to DashboardMetadata fields.
|
|
"""
|
|
query = {
|
|
"columns": ["id", "dashboard_title", "changed_on_utc", "published"]
|
|
}
|
|
_, dashboards = self.get_dashboards(query=query)
|
|
|
|
# Map fields to DashboardMetadata schema
|
|
result = []
|
|
for dash in dashboards:
|
|
result.append({
|
|
"id": dash.get("id"),
|
|
"title": dash.get("dashboard_title"),
|
|
"last_modified": dash.get("changed_on_utc"),
|
|
"status": "published" if dash.get("published") else "draft"
|
|
})
|
|
return result
|
|
# [/DEF:SupersetClient.get_dashboards_summary]
|
|
|
|
# [/DEF:SupersetClient]
|
|
|
|
# [/DEF:backend.src.core.superset_client]
|