__doc__ = """
This FastAPI app uses gradio components with SQL code input
and HTML table output. The query is executed using DuckDB.
The query results are shown in an iframe where the table
is styled and made interactive using Datatables.net scripts.
"""
import gradio as gr
import pandas as pd
from fastapi import FastAPI
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.middleware.cors import CORSMiddleware
from os import getenv
from sql import Q
from itables import options as itoptions, to_html_datatable
itoptions.classes = "display compact cell-border"
itoptions.column_filters = "footer"
itoptions.maxBytes = 2**20
itoptions.layout = {
"top": {
"buttons": ["copyHtml5", "csvHtml5", "pageLength"],
},
"top2": "search",
"top3": "info",
"topStart": None,
"topEnd": None,
"bottom": "paging",
"bottomStart": None,
"bottomEnd": None,
}
CSS = """
#resultContainer {
height: max(880px, calc(90vh - 40px));
box-shadow: var(--block-shadow) !important;
border-width: var(--block-border-width) !important;
border-color: var(--block-border-color) !important;
border-radius: var(--block-radius) !important;
background: var(--block-background-fill) !important;
}
"""
HEAD = """
"""
RESULT_TEMPLATE = f"""
{HEAD}
{{datatable}}
Formatted query:
{{q}}
ran at {{ran}}
"""
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["GET", "POST"],
)
@app.get("/q")
def retrieve_query(query_id: str|None = None, alias: str|None = None):
"""Endpoint for retrieving saved SQL queries."""
q = Q.from_history(query_id=query_id, alias=alias)
return HTMLResponse(content=run_query(q))
def run_query(q: Q, save=True, html_template=RESULT_TEMPLATE):
try:
df = q.df(save=save, _raise=True)
except Exception as e:
df = pd.DataFrame({"error": [str(e)]})
result_datatable = to_html_datatable(df)
html = html_template.format(
datatable=result_datatable,
q=q,
ran=q.end.datetime.strftime("%F %T")
)
return html
def query_from_request(sql_input, definitions, request: gr.Request):
"""Process query from input block or from initial request."""
host = change_hf_host(request)
if not sql_input and not definitions:
# landing page or saved query
url_query_params = dict(request.query_params)
query_id = url_query_params.get("q")
alias = url_query_params.get("alias")
if not query_id and not alias:
alias = "example1"
q = Q.from_history(query_id=query_id, alias=alias)
iframe_src = f"/q?query_id={q.source_id}"
else:
# new query - run button was pressed, register unexecuted query
q = Q.from_template_and_definitions(sql_input, definitions)
q.save()
query_id = q.id
alias = q.alias
hf_space_host = None #getenv("SPACE_HOST")
if hf_space_host:
iframe_src = f"https://{hf_space_host}/q?query_id={query_id}"
else:
iframe_src = f"/q?query_id={query_id}"
result = f"""
"""
sql_input = q.template
definitions = f"{q.definitions}"
editor_url = "".join([
f"http://{host}/?",
f"q={query_id}" if query_id else "",
f"&alias={alias}" if alias else "",
])
result_url = f"http://{host}{iframe_src}"
return (sql_input, definitions, result, editor_url, result_url)
with gr.Blocks(
title="Gradio DuckDB Editor",
theme=gr.themes.Soft(),
css=CSS,
) as gradio_sql_interface:
with gr.Row():
with gr.Column(scale=2, min_width=480):
caption1 = gr.Markdown("# SQL Editor\nClick buttons below to see examples")
with gr.Row(variant="compact"):
ex1_button = gr.Button("Variables", link="/?alias=example1", min_width=100)
ex2_button = gr.Button("URL", link="/?alias=example2", min_width=100)
ex3_button = gr.Button("Local File", link="/?alias=example3", min_width=100)
ex4_button = gr.Button("Bad Query", link="/?alias=bad_example", min_width=100)
definitions = gr.Code(label="Definitions", lines=2, interactive=True)
sql_input = gr.Code(label="SQL Query", language="sql", lines=25, interactive=True)
run_button = gr.Button("run", variant="primary")
editor_url = gr.Code(label="Share Editor URL", lines=1)
result_url = gr.Code(label="Share Query Results URL", lines=1)
with gr.Column(scale=3, min_width=540):
caption2 = gr.Markdown("# RESULTS")
result = gr.HTML(elem_classes="block")
magic = dict(
fn=query_from_request,
inputs=[sql_input, definitions],
outputs=[sql_input, definitions, result, editor_url, result_url]
)
run_button.click(**magic)
gradio_sql_interface.load(**magic)
def change_hf_host(request: gr.Request):
"""Access endpoints that hidden from default HF app landing page."""
_host = request.headers.get("Host")
if "huggingface.co/spaces" in _host:
split_url = _host.rsplit("/", maxsplit=2)
hf_user, hf_space = split_url[1], split_url[2]
host = f"https://{hf_user}-{hf_space}.hf.space"
else:
host = _host
return host
app = gr.mount_gradio_app(app, gradio_sql_interface, path="/")
# @app.get("/")
# # def redirect_hack():
# # return RedirectResponse("/redirecting")
# # @app.get("/redirecting")
# def redirect_to_example():
# hf_space_host = getenv("SPACE_HOST")
# if hf_space_host:
# return RedirectResponse(f"https://{hf_space_host}/sql?alias=example1")
# else:
# return RedirectResponse("/sql?alias=example1")
# return RedirectResponse("/sql?alias=example1")