in superset/commands/explore/get.py [0:0]
def run(self) -> Optional[dict[str, Any]]: # noqa: C901
initial_form_data = {}
if self._permalink_key is not None:
command = GetExplorePermalinkCommand(self._permalink_key)
permalink_value = command.run()
if not permalink_value:
raise ExplorePermalinkGetFailedError()
state = permalink_value["state"]
initial_form_data = state["formData"]
url_params = state.get("urlParams")
if url_params:
initial_form_data["url_params"] = dict(url_params)
elif self._form_data_key:
parameters = FormDataCommandParameters(key=self._form_data_key)
value = GetFormDataCommand(parameters).run()
initial_form_data = json.loads(value) if value else {}
message = None
if not initial_form_data:
if self._slice_id:
initial_form_data["slice_id"] = self._slice_id
if self._form_data_key:
message = _(
"Form data not found in cache, reverting to chart metadata."
)
elif self._datasource_id:
initial_form_data["datasource"] = (
f"{self._datasource_id}__{self._datasource_type}"
)
if self._form_data_key:
message = _(
"Form data not found in cache, reverting to dataset metadata."
)
form_data, slc = get_form_data(
slice_id=self._slice_id,
use_slice_data=True,
initial_form_data=initial_form_data,
)
try:
self._datasource_id, self._datasource_type = get_datasource_info(
self._datasource_id, self._datasource_type, form_data
)
except SupersetException:
self._datasource_id = None
# fallback unknown datasource to table type
self._datasource_type = SqlaTable.type
datasource: Optional[BaseDatasource] = None
if self._datasource_id is not None:
with contextlib.suppress(DatasourceNotFound):
datasource = DatasourceDAO.get_datasource(
cast(str, self._datasource_type), self._datasource_id
)
datasource_name = _("[Missing Dataset]")
if datasource:
datasource_name = datasource.name
security_manager.raise_for_access(datasource=datasource)
viz_type = form_data.get("viz_type")
if not viz_type and datasource and datasource.default_endpoint:
raise WrongEndpointError(redirect=datasource.default_endpoint)
form_data["datasource"] = (
str(self._datasource_id) + "__" + cast(str, self._datasource_type)
)
# On explore, merge legacy/extra filters and URL params into the form data
utils.convert_legacy_filters_into_adhoc(form_data)
utils.merge_extra_filters(form_data)
utils.merge_request_params(form_data, request.args)
# TODO: this is a dummy placeholder - should be refactored to being just `None`
datasource_data: dict[str, Any] = {
"type": self._datasource_type,
"name": datasource_name,
"columns": [],
"metrics": [],
"database": {"id": 0, "backend": ""},
}
try:
if datasource:
datasource_data = datasource.data
except SupersetException as ex:
message = ex.message
except SQLAlchemyError:
message = "SQLAlchemy error"
metadata = None
if slc:
metadata = {
"created_on_humanized": slc.created_on_humanized,
"changed_on_humanized": slc.changed_on_humanized,
"owners": [owner.get_full_name() for owner in slc.owners],
"dashboards": [
{"id": dashboard.id, "dashboard_title": dashboard.dashboard_title}
for dashboard in slc.dashboards
],
}
if slc.created_by:
metadata["created_by"] = slc.created_by.get_full_name()
if slc.changed_by:
metadata["changed_by"] = slc.changed_by.get_full_name()
return {
"dataset": sanitize_datasource_data(datasource_data),
"form_data": form_data,
"slice": slc.data if slc else None,
"message": message,
"metadata": metadata,
}