busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(superset-tool): Correct API response for databases

Browse Source
This commit is contained in:
busya
2025-12-16 10:17:14 +03:00
parent e0e77329bf
commit b735ceb1b0
3 changed files with 17 additions and 774 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
superset_tool/client.py
View File

@@ -83,7 +83,9 @@ class SupersetClient:
def get_dashboards(self, query: Optional[Dict] = None) -> Tuple[int, List[Dict]]:
assert self.network, "[get_dashboards][PRE] Network client must be initialized."
self.logger.info("[get_dashboards][Enter] Fetching dashboards.")
validated_query = self._validate_query_params(query)
validated_query = self._validate_query_params(query or {})
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/",
@@ -306,7 +308,7 @@ class SupersetClient:
# @RETURN: Dict - Валидированные параметры.
def _validate_query_params(self, query: Optional[Dict]) -> Dict:
assert query is None or isinstance(query, dict), "[_validate_query_params][PRE] query must be a dictionary or None."
base_query = {"columns": ["slug", "id", "changed_on_utc", "dashboard_title", "published"], "page": 0, "page_size": 1000}
base_query = {"page": 0, "page_size": 1000}
return {**base_query, **(query or {})}
# [/DEF:SupersetClient._validate_query_params]
@@ -369,6 +371,7 @@ class SupersetClient:
assert self.network, "[get_datasets][PRE] Network client must be initialized."
self.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/",
@@ -390,7 +393,9 @@ class SupersetClient:
def get_databases(self, query: Optional[Dict] = None) -> Tuple[int, List[Dict]]:
assert self.network, "[get_databases][PRE] Network client must be initialized."
self.logger.info("[get_databases][Enter] Fetching databases.")
validated_query = self._validate_query_params(query)
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/",