-
Notifications
You must be signed in to change notification settings - Fork 3
/
quack.py
36 lines (29 loc) · 1.03 KB
/
quack.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from pydantic import BaseModel
from fastapi.responses import JSONResponse
from modal import Image, App, web_endpoint
import json
from datetime import date, datetime
image = Image.debian_slim().pip_install("duckdb==1.0.0")
app = App(name="quack")
class Query(BaseModel):
sql: str
def json_serial(obj):
if isinstance(obj, (datetime, date)):
return obj.isoformat()
raise TypeError(f"Type {type(obj)} not serializable")
@app.function(image=image, concurrency_limit=5)
@web_endpoint(method="POST")
def query_duckdb(query: Query):
import duckdb
try:
conn = duckdb.connect()
result = conn.execute(query.sql).fetchall()
column_names = [desc[0] for desc in conn.description]
formatted_result = [dict(zip(column_names, row)) for row in result]
conn.close()
json_compatible_result = json.loads(
json.dumps(formatted_result, default=json_serial)
)
return JSONResponse(content={"result": json_compatible_result})
except Exception as e:
return JSONResponse(content={"error": str(e)}, status_code=400)