The UDF and the code in app.py is not aligned. retrieve_sl_metadata() function does not take any input variables.
But is used as:
sql = f"select retrieve_sl_metadata('{host}', {environment_id}, '{token}')" in app.py
UDF:
create or replace function retrieve_sl_metadata()
returns object
language python
runtime_version = 3.9
handler = 'main'
external_access_integrations = (dbt_cloud_semantic_layer_integration)
packages = ('requests')
secrets = ('cred' = dbt_cloud_service_token)
as
$$
from typing import Dict
import _snowflake
import requests
query = """
query GetMetrics($environmentId: BigInt!) {
metrics(environmentId: $environmentId) {
description
name
queryableGranularities
type
dimensions {
description
name
type
}
}
}
in app.py```
def prepare_app():
host = st.session_state.get("host")
environment_id = st.session_state.get("environment_id")
token = st.session_state.get("token")
if any([is_none(i) for i in [host, environment_id]]):
st.error("Host and environment ID are required inputs")
st.stop()
sql = f"select retrieve_sl_metadata('{host}', {environment_id}, '{token}')"
with st.spinner("Retrieving metadata..."):
response = session.sql(sql).collect()
data = json.loads(response[0][0])
try:
metrics = data["data"]["metrics"]
except TypeError:
# `data` is None and there may be an error
try:
error = data["errors"][0]["message"]
st.error(error)
except (KeyError, TypeError):
st.warning(
"No metrics returned. Ensure your project has metrics defined "
"and a production job has been run successfully."
)
else:
st.session_state.metric_dict = {m["name"]: m for m in metrics}
st.session_state.dimension_dict = {
dim["name"]: dim for metric in metrics for dim in metric["dimensions"]
}
st.session_state.metrics = ", ".join(list(st.session_state.metric_dict.keys()))
st.session_state.dimensions = ", ".join(list(st.session_state.dimension_dict.keys()))
for metric in st.session_state.metric_dict:
st.session_state.metric_dict[metric]["dimensions"] = [
d["name"]
for d in st.session_state.metric_dict[metric]["dimensions"]
]
if not st.session_state.metric_dict:
# Query worked, but nothing returned
st.warning(
"No Metrics returned! Ensure your project has metrics defined "
"and a production job has been run successfully."
)
else:
st.success("Success! Start asking questions!")
st.experimental_rerun()
The UDF and the code in app.py is not aligned. retrieve_sl_metadata() function does not take any input variables.
But is used as:
sql = f"select retrieve_sl_metadata('{host}', {environment_id}, '{token}')"
in app.pyUDF:
in app.py```
def prepare_app():