import os
import requests
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
app = FastAPI(title="SQLite Multi-Project on HF Space")
SQLITE_URL = "http://localhost:8000"
class QueryRequest(BaseModel):
sql: str
project: str = "default"
class ExecuteRequest(BaseModel):
sql: str
project: str = "default"
class ProjectRequest(BaseModel):
name: str
description: str = ""
@app.get("/", response_class=HTMLResponse)
def read_root():
return """
SQLite Multi-Project Server
✅ SQLite Multi-Project Server
Persistent Storage: /data/ (organized by project)
API Docs: Swagger UI
Project Management
GET /api/v1/projects - List all projectsPOST /api/v1/projects - Create new projectDELETE /api/v1/projects/{name} - Delete project
Database Operations
GET /api/v1/tables?project=myproject - List tablesGET /api/v1/schema?project=myproject - Get schemaPOST /api/v1/query - Execute SELECTPOST /api/v1/execute - Execute INSERT/UPDATE/DELETE
Example: Multiple Projects
📁 /data/
├── project1.db (users, orders tables)
├── project2.db (products, inventory tables)
└── project3.db (logs, analytics tables)
"""
@app.get("/portal", response_class=HTMLResponse)
def portal():
return """
SQLite Portal
SQLite Portal
Loading database info...
SQL Editor
"""
@app.get("/health")
def health_check():
try:
response = requests.get(f"{SQLITE_URL}/api/v1/heartbeat", timeout=5)
if response.status_code == 200:
return {"status": "healthy", "sqlite": "up"}
else:
return {"status": "unhealthy", "detail": response.text}
except Exception as e:
return {"status": "error", "detail": str(e)}
@app.get("/api/v1/heartbeat")
def heartbeat():
try:
response = requests.get(f"{SQLITE_URL}/api/v1/heartbeat", timeout=5)
return response.json()
except Exception as e:
raise HTTPException(status_code=500, detail=f"SQLite server error: {str(e)}")
@app.get("/api/v1/projects")
def list_projects():
try:
response = requests.get(f"{SQLITE_URL}/api/v1/projects", timeout=5)
if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail=response.text)
return response.json()
except Exception as e:
raise HTTPException(status_code=500, detail=f"Projects error: {str(e)}")
@app.post("/api/v1/projects")
def create_project(request: ProjectRequest):
try:
response = requests.post(f"{SQLITE_URL}/api/v1/projects", json={"name": request.name, "description": request.description}, timeout=5)
if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail=response.text)
return response.json()
except Exception as e:
raise HTTPException(status_code=500, detail=f"Create project error: {str(e)}")
@app.delete("/api/v1/projects/{project_name}")
def delete_project(project_name: str):
try:
response = requests.delete(f"{SQLITE_URL}/api/v1/projects/{project_name}", timeout=5)
if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail=response.text)
return response.json()
except Exception as e:
raise HTTPException(status_code=500, detail=f"Delete project error: {str(e)}")
@app.post("/api/v1/query")
def query(request: QueryRequest):
try:
response = requests.post(f"{SQLITE_URL}/api/v1/query", json={"sql": request.sql, "project": request.project}, timeout=10)
if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail=response.text)
return response.json()
except Exception as e:
raise HTTPException(status_code=500, detail=f"Query error: {str(e)}")
@app.post("/api/v1/execute")
def execute(request: ExecuteRequest):
try:
response = requests.post(f"{SQLITE_URL}/api/v1/execute", json={"sql": request.sql, "project": request.project}, timeout=10)
if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail=response.text)
return response.json()
except Exception as e:
raise HTTPException(status_code=500, detail=f"Execute error: {str(e)}")
@app.get("/api/v1/tables")
def list_tables(project: str = Query("default")):
try:
response = requests.get(f"{SQLITE_URL}/api/v1/tables?project={project}", timeout=5)
if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail=response.text)
return response.json()
except Exception as e:
raise HTTPException(status_code=500, detail=f"Tables error: {str(e)}")
@app.get("/api/v1/schema")
def get_schema(project: str = Query("default")):
try:
response = requests.get(f"{SQLITE_URL}/api/v1/schema?project={project}", timeout=5)
if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail=response.text)
return response.json()
except Exception as e:
raise HTTPException(status_code=500, detail=f"Schema error: {str(e)}")
if __name__ == "__main__":
import uvicorn
port = int(os.getenv("PORT", 7860))
uvicorn.run(app, host="0.0.0.0", port=port)