Further extending the work from @Ferris and @kenske, when using localhost (as this was a local script), I found all the session cookies and redirects kept (trying) to go to https on localhost, which I didn't want. Below is my final solution that allows updating the /embedded
endpoint, as well as reading the list of dashboards etc. I'm guessing with the the use of the CSRF token this will not work for long running scripts, but for a one shot script, this is working well:
import json
import logging
import os
import re
import requests
logging.basicConfig(level=logging.INFO)
host = "localhost:8088"
api_base = f"http://{host}/api/v1"
username = "admin"
password = os.getenv("DEFAULT_ADMIN_PASSWORD")
dashboard_configs = []
def get_superset_session() -> requests.Session:
"""
A requests.session with the cookies set such that the work against the API
"""
# set up session for auth
session = requests.Session()
session.headers.update(
{
# With "ENABLE_PROXY_FIX = True" in superset_config.py, we need to set this
# header to make superset think we are using https
"X-Forwarded-Proto": "https",
"Accept": "application/json",
# CRITICAL: Without this, we don't get the correct real session cookie
"Referer": f"https://{host}/login/",
}
)
login_form = session.get(
f"http://{host}/login/",
# Disable redirects as it'll redirect to https:// on localhost, which won't work
allow_redirects=False,
)
# Force cookies to be http so requests still sends them
for cookie in session.cookies:
cookie.secure = False
# get Cross-Site Request Forgery protection token
match = re.search(
r'<input[^>]*id="csrf_token"[^>]*value="([^"]+)"', login_form.text
)
if match:
csrf_token = match.group(1)
else:
raise Exception("CSRF token not found")
data = {"username": username, "password": password, "csrf_token": csrf_token}
# login the given session
session.post(
f"http://{host}/login/",
data=data,
allow_redirects=False,
)
for cookie in session.cookies:
cookie.secure = False
# Set the CSRF token header, without this, some POSTs don't work, eg: "embeded"
response = session.get(
f"{api_base}/security/csrf_token/",
headers={
"Accept": "application/json",
},
)
csrf_token = response.json()["result"]
session.headers.update(
{
"X-CSRFToken": csrf_token,
}
)
return session
def generate_embed_uuid(session: requests.Session, dashboard_id: int):
"""
Generate an embed UUID for the given dashboard ID
"""
response = session.post(
f"{api_base}/dashboard/{dashboard_id}/embedded",
json={"allowed_domains": []},
)
response.raise_for_status()
return response.json().get("result", {}).get("uuid")
def main():
session = get_superset_session()
dashboard_query = {
"columns": ["dashboard_title", "id"],
}
response = session.get(
f"{api_base}/dashboard/",
params={"q": json.dumps(dashboard_query)},
)
dashboards = response.json()
for dashboard in dashboards["result"]:
dashboard_id = dashboard["id"]
dashboard_title = dashboard["dashboard_title"]
response = session.get(f"{api_base}/dashboard/{dashboard['id']}/embedded")
embed_uuid = response.json().get("result", {}).get("uuid")
if not embed_uuid:
print(f"Generating embed UUID for {dashboard_title} ({dashboard_id})...")
embed_uuid = generate_embed_uuid(session, dashboard_id)
embed_config = {
"dashboard_id": dashboard_id,
"dashboard_title": dashboard_title,
"embed_uuid": embed_uuid,
}
print("Embed Config:", embed_config)
dashboard_configs.append(embed_config)
print(dashboard_configs)
if __name__ == "__main__":
main()