Files

192 lines
5.5 KiB
Python
Raw Permalink Normal View History

2026-02-03 21:58:25 +01:00
"""
=============================================================================
PROJECTS ROUTER
=============================================================================
Example CRUD endpoints for managing projects.
This demonstrates:
- Supabase database integration
- CRUD operations (Create, Read, Update, Delete)
- Request validation with Pydantic
- Error handling patterns
You can adapt this pattern for any resource in your app.
"""
from fastapi import APIRouter, HTTPException, Header
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
# Import Supabase client
from services.database import get_supabase_client
router = APIRouter()
# =============================================================================
# REQUEST/RESPONSE MODELS
# =============================================================================
class ProjectCreate(BaseModel):
"""Data required to create a new project."""
name: str = Field(..., min_length=1, max_length=100)
description: Optional[str] = Field(default="", max_length=1000)
class ProjectUpdate(BaseModel):
"""Data that can be updated on a project."""
name: Optional[str] = Field(None, min_length=1, max_length=100)
description: Optional[str] = Field(None, max_length=1000)
class Project(BaseModel):
"""Full project model (database representation)."""
id: str
name: str
description: str
user_id: str
created_at: datetime
# =============================================================================
# ENDPOINTS
# =============================================================================
@router.get("")
async def list_projects(
authorization: str = Header(None, description="Bearer token from Supabase")
):
"""
List all projects for the authenticated user.
Requires authentication - pass the Supabase access token
in the Authorization header.
"""
if not authorization:
raise HTTPException(status_code=401, detail="Authorization required")
try:
# Get Supabase client
supabase = get_supabase_client()
# Extract user from token (Supabase handles this)
# In production, you'd verify the JWT properly
# Query projects
response = supabase.table("projects").select("*").execute()
return response.data
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("", response_model=Project)
async def create_project(
project: ProjectCreate,
authorization: str = Header(None),
):
"""
Create a new project.
Example:
POST /api/projects
{"name": "My AI App", "description": "An awesome AI project"}
"""
if not authorization:
raise HTTPException(status_code=401, detail="Authorization required")
try:
supabase = get_supabase_client()
# In production, extract user_id from JWT
# For demo, we'll use a placeholder
user_id = "demo-user-id"
response = supabase.table("projects").insert({
"name": project.name,
"description": project.description or "",
"user_id": user_id,
}).execute()
return response.data[0]
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{project_id}", response_model=Project)
async def get_project(project_id: str):
"""
Get a specific project by ID.
"""
try:
supabase = get_supabase_client()
response = supabase.table("projects").select("*").eq("id", project_id).single().execute()
if not response.data:
raise HTTPException(status_code=404, detail="Project not found")
return response.data
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.patch("/{project_id}", response_model=Project)
async def update_project(
project_id: str,
project: ProjectUpdate,
authorization: str = Header(None),
):
"""
Update a project.
Only send the fields you want to update.
"""
if not authorization:
raise HTTPException(status_code=401, detail="Authorization required")
try:
supabase = get_supabase_client()
# Build update dict with only provided fields
update_data = project.model_dump(exclude_unset=True)
if not update_data:
raise HTTPException(status_code=400, detail="No fields to update")
response = supabase.table("projects").update(update_data).eq("id", project_id).execute()
if not response.data:
raise HTTPException(status_code=404, detail="Project not found")
return response.data[0]
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.delete("/{project_id}")
async def delete_project(
project_id: str,
authorization: str = Header(None),
):
"""
Delete a project.
"""
if not authorization:
raise HTTPException(status_code=401, detail="Authorization required")
try:
supabase = get_supabase_client()
response = supabase.table("projects").delete().eq("id", project_id).execute()
return {"message": "Project deleted", "id": project_id}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))