task panel
This commit is contained in:
115182
backend/logs/app.log.1
115182
backend/logs/app.log.1
File diff suppressed because it is too large
Load Diff
Binary file not shown.
@@ -187,19 +187,19 @@ async def migrate_dashboards(
|
||||
task_params = {
|
||||
'source_env_id': request.source_env_id,
|
||||
'target_env_id': request.target_env_id,
|
||||
'dashboards': request.dashboard_ids,
|
||||
'selected_ids': request.dashboard_ids,
|
||||
'replace_db_config': request.replace_db_config,
|
||||
'db_mappings': request.db_mappings or {}
|
||||
}
|
||||
|
||||
task_id = await task_manager.create_task(
|
||||
task_obj = await task_manager.create_task(
|
||||
plugin_id='superset-migration',
|
||||
params=task_params
|
||||
)
|
||||
|
||||
logger.info(f"[migrate_dashboards][Coherence:OK] Migration task created: {task_id} for {len(request.dashboard_ids)} dashboards")
|
||||
logger.info(f"[migrate_dashboards][Coherence:OK] Migration task created: {task_obj.id} for {len(request.dashboard_ids)} dashboards")
|
||||
|
||||
return TaskResponse(task_id=str(task_id))
|
||||
return TaskResponse(task_id=str(task_obj.id))
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[migrate_dashboards][Coherence:Failed] Failed to create migration task: {e}")
|
||||
@@ -254,14 +254,14 @@ async def backup_dashboards(
|
||||
'schedule': request.schedule
|
||||
}
|
||||
|
||||
task_id = await task_manager.create_task(
|
||||
task_obj = await task_manager.create_task(
|
||||
plugin_id='superset-backup',
|
||||
params=task_params
|
||||
)
|
||||
|
||||
logger.info(f"[backup_dashboards][Coherence:OK] Backup task created: {task_id} for {len(request.dashboard_ids)} dashboards")
|
||||
logger.info(f"[backup_dashboards][Coherence:OK] Backup task created: {task_obj.id} for {len(request.dashboard_ids)} dashboards")
|
||||
|
||||
return TaskResponse(task_id=str(task_id))
|
||||
return TaskResponse(task_id=str(task_obj.id))
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[backup_dashboards][Coherence:Failed] Failed to create backup task: {e}")
|
||||
@@ -272,6 +272,8 @@ async def backup_dashboards(
|
||||
class DatabaseMapping(BaseModel):
|
||||
source_db: str
|
||||
target_db: str
|
||||
source_db_uuid: Optional[str] = None
|
||||
target_db_uuid: Optional[str] = None
|
||||
confidence: float
|
||||
# [/DEF:DatabaseMapping:DataClass]
|
||||
|
||||
@@ -306,6 +308,8 @@ async def get_database_mappings(
|
||||
DatabaseMapping(
|
||||
source_db=s.get('source_db', ''),
|
||||
target_db=s.get('target_db', ''),
|
||||
source_db_uuid=s.get('source_db_uuid'),
|
||||
target_db_uuid=s.get('target_db_uuid'),
|
||||
confidence=s.get('confidence', 0.0)
|
||||
)
|
||||
for s in suggestions
|
||||
|
||||
@@ -270,21 +270,21 @@ async def map_columns(
|
||||
try:
|
||||
# Create mapping task
|
||||
task_params = {
|
||||
'env_id': request.env_id,
|
||||
'datasets': request.dataset_ids,
|
||||
'source_type': request.source_type,
|
||||
'env': request.env_id,
|
||||
'dataset_id': request.dataset_ids[0] if request.dataset_ids else None,
|
||||
'source': request.source_type,
|
||||
'connection_id': request.connection_id,
|
||||
'file_data': request.file_data
|
||||
}
|
||||
|
||||
task_id = await task_manager.create_task(
|
||||
task_obj = await task_manager.create_task(
|
||||
plugin_id='dataset-mapper',
|
||||
params=task_params
|
||||
)
|
||||
|
||||
logger.info(f"[map_columns][Coherence:OK] Mapping task created: {task_id} for {len(request.dataset_ids)} datasets")
|
||||
logger.info(f"[map_columns][Coherence:OK] Mapping task created: {task_obj.id} for {len(request.dataset_ids)} datasets")
|
||||
|
||||
return TaskResponse(task_id=str(task_id))
|
||||
return TaskResponse(task_id=str(task_obj.id))
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[map_columns][Coherence:Failed] Failed to create mapping task: {e}")
|
||||
@@ -334,20 +334,20 @@ async def generate_docs(
|
||||
try:
|
||||
# Create documentation generation task
|
||||
task_params = {
|
||||
'env_id': request.env_id,
|
||||
'datasets': request.dataset_ids,
|
||||
'llm_provider': request.llm_provider,
|
||||
'environment_id': request.env_id,
|
||||
'dataset_id': str(request.dataset_ids[0]) if request.dataset_ids else None,
|
||||
'provider_id': request.llm_provider,
|
||||
'options': request.options or {}
|
||||
}
|
||||
|
||||
task_id = await task_manager.create_task(
|
||||
task_obj = await task_manager.create_task(
|
||||
plugin_id='llm_documentation',
|
||||
params=task_params
|
||||
)
|
||||
|
||||
logger.info(f"[generate_docs][Coherence:OK] Documentation generation task created: {task_id} for {len(request.dataset_ids)} datasets")
|
||||
logger.info(f"[generate_docs][Coherence:OK] Documentation generation task created: {task_obj.id} for {len(request.dataset_ids)} datasets")
|
||||
|
||||
return TaskResponse(task_id=str(task_id))
|
||||
return TaskResponse(task_id=str(task_obj.id))
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[generate_docs][Coherence:Failed] Failed to create documentation generation task: {e}")
|
||||
|
||||
@@ -31,6 +31,7 @@ class MappingCreate(BaseModel):
|
||||
target_db_uuid: str
|
||||
source_db_name: str
|
||||
target_db_name: str
|
||||
engine: Optional[str] = None
|
||||
# [/DEF:MappingCreate:DataClass]
|
||||
|
||||
# [DEF:MappingResponse:DataClass]
|
||||
@@ -42,6 +43,7 @@ class MappingResponse(BaseModel):
|
||||
target_db_uuid: str
|
||||
source_db_name: str
|
||||
target_db_name: str
|
||||
engine: Optional[str] = None
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
@@ -94,6 +96,7 @@ async def create_mapping(
|
||||
if existing:
|
||||
existing.target_db_uuid = mapping.target_db_uuid
|
||||
existing.target_db_name = mapping.target_db_name
|
||||
existing.engine = mapping.engine
|
||||
db.commit()
|
||||
db.refresh(existing)
|
||||
return existing
|
||||
|
||||
@@ -87,11 +87,11 @@ class SupersetClient:
|
||||
if 'columns' not in validated_query:
|
||||
validated_query['columns'] = ["slug", "id", "changed_on_utc", "dashboard_title", "published"]
|
||||
|
||||
total_count = self._fetch_total_object_count(endpoint="/dashboard/")
|
||||
paginated_data = self._fetch_all_pages(
|
||||
endpoint="/dashboard/",
|
||||
pagination_options={"base_query": validated_query, "total_count": total_count, "results_field": "result"},
|
||||
pagination_options={"base_query": validated_query, "results_field": "result"},
|
||||
)
|
||||
total_count = len(paginated_data)
|
||||
app_logger.info("[get_dashboards][Exit] Found %d dashboards.", total_count)
|
||||
return total_count, paginated_data
|
||||
# [/DEF:get_dashboards:Function]
|
||||
@@ -203,11 +203,11 @@ class SupersetClient:
|
||||
app_logger.info("[get_datasets][Enter] Fetching datasets.")
|
||||
validated_query = self._validate_query_params(query)
|
||||
|
||||
total_count = self._fetch_total_object_count(endpoint="/dataset/")
|
||||
paginated_data = self._fetch_all_pages(
|
||||
endpoint="/dataset/",
|
||||
pagination_options={"base_query": validated_query, "total_count": total_count, "results_field": "result"},
|
||||
pagination_options={"base_query": validated_query, "results_field": "result"},
|
||||
)
|
||||
total_count = len(paginated_data)
|
||||
app_logger.info("[get_datasets][Exit] Found %d datasets.", total_count)
|
||||
return total_count, paginated_data
|
||||
# [/DEF:get_datasets:Function]
|
||||
@@ -370,11 +370,12 @@ class SupersetClient:
|
||||
validated_query = self._validate_query_params(query or {})
|
||||
if 'columns' not in validated_query:
|
||||
validated_query['columns'] = []
|
||||
total_count = self._fetch_total_object_count(endpoint="/database/")
|
||||
|
||||
paginated_data = self._fetch_all_pages(
|
||||
endpoint="/database/",
|
||||
pagination_options={"base_query": validated_query, "total_count": total_count, "results_field": "result"},
|
||||
pagination_options={"base_query": validated_query, "results_field": "result"},
|
||||
)
|
||||
total_count = len(paginated_data)
|
||||
app_logger.info("[get_databases][Exit] Found %d databases.", total_count)
|
||||
return total_count, paginated_data
|
||||
# [/DEF:get_databases:Function]
|
||||
|
||||
@@ -355,20 +355,40 @@ class APIClient:
|
||||
# @PURPOSE: Автоматически собирает данные со всех страниц пагинированного эндпоинта.
|
||||
# @PARAM: endpoint (str) - Эндпоинт.
|
||||
# @PARAM: pagination_options (Dict[str, Any]) - Опции пагинации.
|
||||
# @PRE: pagination_options must contain 'base_query', 'total_count', 'results_field'.
|
||||
# @PRE: pagination_options must contain 'base_query', 'results_field'. 'total_count' is optional.
|
||||
# @POST: Returns all items across all pages.
|
||||
# @RETURN: List[Any] - Список данных.
|
||||
def fetch_paginated_data(self, endpoint: str, pagination_options: Dict[str, Any]) -> List[Any]:
|
||||
with belief_scope("fetch_paginated_data"):
|
||||
base_query, total_count = pagination_options["base_query"], pagination_options["total_count"]
|
||||
results_field, page_size = pagination_options["results_field"], base_query.get('page_size')
|
||||
assert page_size and page_size > 0, "'page_size' must be a positive number."
|
||||
base_query = pagination_options["base_query"]
|
||||
total_count = pagination_options.get("total_count")
|
||||
|
||||
results_field = pagination_options["results_field"]
|
||||
count_field = pagination_options.get("count_field", "count")
|
||||
page_size = base_query.get('page_size', 1000)
|
||||
assert page_size > 0, "'page_size' must be a positive number."
|
||||
|
||||
results = []
|
||||
for page in range((total_count + page_size - 1) // page_size):
|
||||
page = 0
|
||||
|
||||
# Fetch first page to get data and total count if not provided
|
||||
query = {**base_query, 'page': page}
|
||||
response_json = cast(Dict[str, Any], self.request("GET", endpoint, params={"q": json.dumps(query)}))
|
||||
|
||||
first_page_results = response_json.get(results_field, [])
|
||||
results.extend(first_page_results)
|
||||
|
||||
if total_count is None:
|
||||
total_count = response_json.get(count_field, len(first_page_results))
|
||||
app_logger.debug(f"[fetch_paginated_data][State] Total count resolved from first page: {total_count}")
|
||||
|
||||
# Fetch remaining pages
|
||||
total_pages = (total_count + page_size - 1) // page_size
|
||||
for page in range(1, total_pages):
|
||||
query = {**base_query, 'page': page}
|
||||
response_json = cast(Dict[str, Any], self.request("GET", endpoint, params={"q": json.dumps(query)}))
|
||||
results.extend(response_json.get(results_field, []))
|
||||
|
||||
return results
|
||||
# [/DEF:fetch_paginated_data:Function]
|
||||
|
||||
|
||||
@@ -219,22 +219,29 @@ class MigrationPlugin(PluginBase):
|
||||
log.warning("No dashboards found matching criteria.")
|
||||
return
|
||||
|
||||
# Fetch mappings from database
|
||||
db_mapping = {}
|
||||
# Get mappings from params
|
||||
db_mapping = params.get("db_mappings", {})
|
||||
if not isinstance(db_mapping, dict):
|
||||
db_mapping = {}
|
||||
|
||||
# Fetch additional mappings from database if requested
|
||||
if replace_db_config:
|
||||
db = SessionLocal()
|
||||
try:
|
||||
# Find environment IDs by name
|
||||
src_env = db.query(Environment).filter(Environment.name == from_env_name).first()
|
||||
tgt_env = db.query(Environment).filter(Environment.name == to_env_name).first()
|
||||
src_env_db = db.query(Environment).filter(Environment.name == from_env_name).first()
|
||||
tgt_env_db = db.query(Environment).filter(Environment.name == to_env_name).first()
|
||||
|
||||
if src_env and tgt_env:
|
||||
mappings = db.query(DatabaseMapping).filter(
|
||||
DatabaseMapping.source_env_id == src_env.id,
|
||||
DatabaseMapping.target_env_id == tgt_env.id
|
||||
if src_env_db and tgt_env_db:
|
||||
stored_mappings = db.query(DatabaseMapping).filter(
|
||||
DatabaseMapping.source_env_id == src_env_db.id,
|
||||
DatabaseMapping.target_env_id == tgt_env_db.id
|
||||
).all()
|
||||
db_mapping = {m.source_db_uuid: m.target_db_uuid for m in mappings}
|
||||
log.info(f"Loaded {len(db_mapping)} database mappings.")
|
||||
# Provided mappings override stored ones
|
||||
stored_map_dict = {m.source_db_uuid: m.target_db_uuid for m in stored_mappings}
|
||||
stored_map_dict.update(db_mapping)
|
||||
db_mapping = stored_map_dict
|
||||
log.info(f"Loaded {len(stored_mappings)} database mappings from database.")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
BIN
backend/tasks.db
BIN
backend/tasks.db
Binary file not shown.
Reference in New Issue
Block a user