File size: 5,771 Bytes
58da524 | 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 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 | # main.py
from fastapi import FastAPI, HTTPException, APIRouter
from fastapi.middleware.cors import CORSMiddleware
import crud
import schemas
from Idiom_Id_Generator import generate_id
from dotenv import load_dotenv
import os
from pydantic import BaseModel
from typing import List, Dict, Optional
load_dotenv()
app = FastAPI(title="Idioms API - Supabase REST")
allowed_origins = [
"http://localhost",
"http://localhost:3000",
"http://localhost:8000",
"https://idiomator.vercel.app",
"https://www.idiomator.vercel.app",
"https://idiomator.com",
"https://www.idiomator.com",
]
app.add_middleware(
CORSMiddleware,
allow_origins= allowed_origins,#allowed_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/idioms", response_model=list[schemas.IdiomResponse])
async def read_idioms(skip: int = 0, limit: int = 50):
try:
data = await crud.get_idioms(skip=skip, limit=limit)
return data
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/idioms", response_model=schemas.IdiomResponse)
async def create_idiom(idiom: schemas.IdiomCreate):
try:
# keep your ID generation
idiom_id = generate_id(idiom.language, idiom.dialect)
idiom_dict = idiom.dict()
idiom_dict["id"] = idiom_id
created = await crud.create_idiom(idiom_dict)
return created
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/idioms/all_ids", response_model=List[Dict[str, str]])
async def get_all_idioms_route():
try:
idioms = await crud.get_all_idioms()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
return [
{"id": idiom["id"], "idiom": idiom["idiom"], "language": idiom["language"], "updated_at": idiom.get("updated_at")}
for idiom in idioms
]
@app.get("/idioms/search", response_model=List[schemas.IdiomResponse])
async def search_idioms(
q: Optional[str] = None,
language: Optional[str] = None,
skip: int = 0,
limit: int = 50
):
try:
results = await crud.search_idioms(query=q, language=language, skip=skip, limit=limit)
return results
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/idioms/slug/{language}/{slug}", response_model=schemas.IdiomResponse)
async def get_idiom_by_slug(language: str, slug: str):
"""Fetch an idiom by its URL slug + language code.
The slug is the idiom text lowercased with spaces replaced by hyphens."""
try:
idiom = await crud.get_idiom_by_slug(language, slug)
if not idiom:
raise HTTPException(status_code=404, detail="Idiom not found")
if isinstance(idiom.get("validation_count"), int):
idiom["validation_count"] = {"count": idiom["validation_count"]}
return idiom
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/idioms/related/{language}/{idiom_id}")
async def get_related_idioms_route(language: str, idiom_id: str, limit: int = 5):
"""Return up to `limit` other idioms in the same language sharing tokens with the source."""
try:
return await crud.get_related_idioms(language, idiom_id, limit=limit)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/idioms/{idiom_id}", response_model=schemas.IdiomResponse)
async def read_idiom(idiom_id: str):
try:
db_idiom = await crud.get_idiom(idiom_id)
if not db_idiom:
raise HTTPException(status_code=404, detail="Idiom not found")
# Transform validation_count if needed
if isinstance(db_idiom.get("validation_count"), int):
db_idiom["validation_count"] = {"count": db_idiom["validation_count"]}
return db_idiom
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# ---------------------------------------------------------------------------
# REMOVED 2026-07-31 — unauthenticated destructive endpoints.
#
# @app.patch("/idioms/{idiom_id}")
# @app.delete("/idioms/{idiom_id}")
#
# Both were public, took no credential of any kind, and executed against
# Supabase with an `sb_secret_` key that bypasses Row Level Security. Any
# person on the internet could delete or rewrite any idiom with one curl.
# CORS did not protect them — CORS is a browser policy, not a server control.
#
# Neither had a single caller anywhere in idiomator-next/src. If admin writes
# are ever needed, they belong behind an authenticated route, not here.
# ---------------------------------------------------------------------------
@app.get("/ping")
async def ping():
return {"status": "ok"}
@app.get("/health/deep")
async def health_deep():
"""Content-asserting health check for the external uptime monitor.
Performs a real lookup and requires real content back. A plain 200 is not
evidence of anything: this service answered 200 on `/ping` throughout a
seven-week outage in which every dictionary page rendered empty.
"""
try:
idiom = await crud.get_idiom_by_slug("en", "break-the-ice")
except Exception as e:
raise HTTPException(status_code=503, detail=f"lookup failed: {e}")
meanings = (idiom or {}).get("meanings") or []
if not any(m.get("definitions") for m in meanings):
raise HTTPException(
status_code=503,
detail="probe idiom resolved but carries no definitions",
)
return {"status": "IDIOMATOR_DICT_OK", "senses": len(meanings)}
|