wrd_drishti / app /api /routes.py
devarshia5's picture
Upload 21 files
fe7cfe5 verified
Raw
History Blame Contribute Delete
4.18 kB
"""
API Routes β€” Your Application Endpoints
────────────────────────────────────────
This is where your actual API endpoints go.
The admin panel is separate at /admin/*
"""
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import Optional
import uuid
from app.database import get_db
router = APIRouter(prefix="/api", tags=["api"])
# ── Models ──
class UserCreate(BaseModel):
email: str
name: Optional[str] = None
avatar_url: Optional[str] = None
class ProjectCreate(BaseModel):
user_id: str
name: str
description: Optional[str] = None
design_tokens: Optional[dict] = None
class ColorPaletteCreate(BaseModel):
project_id: str
name: str
colors: dict
mood: Optional[str] = None
# ── Users ──
@router.post("/users")
async def create_user(user: UserCreate):
conn = get_db()
user_id = str(uuid.uuid4())
try:
conn.execute(
"INSERT INTO users (id, email, name, avatar_url) VALUES (?, ?, ?, ?)",
(user_id, user.email, user.name, user.avatar_url)
)
conn.commit()
return {"id": user_id, "email": user.email}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@router.get("/users")
async def list_users():
conn = get_db()
rows = conn.execute("SELECT * FROM users ORDER BY created_at DESC").fetchall()
return [dict(r) for r in rows]
@router.get("/users/{user_id}")
async def get_user(user_id: str):
conn = get_db()
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
if not row:
raise HTTPException(status_code=404, detail="User not found")
return dict(row)
# ── Projects ──
@router.post("/projects")
async def create_project(project: ProjectCreate):
conn = get_db()
project_id = str(uuid.uuid4())
try:
conn.execute(
"INSERT INTO projects (id, user_id, name, description, design_tokens) VALUES (?, ?, ?, ?, ?)",
(project_id, project.user_id, project.name, project.description, str(project.design_tokens))
)
conn.commit()
return {"id": project_id, "name": project.name}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@router.get("/projects")
async def list_projects(user_id: Optional[str] = None):
conn = get_db()
if user_id:
rows = conn.execute("SELECT * FROM projects WHERE user_id = ? ORDER BY created_at DESC", (user_id,)).fetchall()
else:
rows = conn.execute("SELECT * FROM projects ORDER BY created_at DESC").fetchall()
return [dict(r) for r in rows]
@router.get("/projects/{project_id}")
async def get_project(project_id: str):
conn = get_db()
row = conn.execute("SELECT * FROM projects WHERE id = ?", (project_id,)).fetchone()
if not row:
raise HTTPException(status_code=404, detail="Project not found")
return dict(row)
# ── Color Palettes ──
@router.post("/palettes")
async def create_palette(palette: ColorPaletteCreate):
conn = get_db()
palette_id = str(uuid.uuid4())
try:
conn.execute(
"INSERT INTO color_palettes (id, project_id, name, colors, mood) VALUES (?, ?, ?, ?, ?)",
(palette_id, palette.project_id, palette.name, str(palette.colors), palette.mood)
)
conn.commit()
return {"id": palette_id, "name": palette.name}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@router.get("/palettes")
async def list_palettes(project_id: Optional[str] = None):
conn = get_db()
if project_id:
rows = conn.execute("SELECT * FROM color_palettes WHERE project_id = ? ORDER BY created_at DESC", (project_id,)).fetchall()
else:
rows = conn.execute("SELECT * FROM color_palettes ORDER BY created_at DESC").fetchall()
return [dict(r) for r in rows]
# ── Health Check ──
@router.get("/health")
async def health_check():
from app.database import integrity_check
return integrity_check()