Spaces:

Justarandomperson
/

Multilingual_Idiom_Extractor

Sleeping

Multilingual_Idiom_Extractor / Backend /Database /main.py

GitHub Actions

Track large files with LFS

58da524 3 months ago

5.77 kB

# 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)}

	# 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)}