src/ddgs_api/api.py

import os
from pathlib import Path
from typing import Optional

from fastapi import Depends, FastAPI, Header, HTTPException, status
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, Field

from ddgs_api.config import ensure_ca_bundle, env_bool, env_int, env_str, load_env
from ddgs_api.search import ddgs_search
from ddgs_api.ui import render_homepage


def require_auth(authorization: Optional[str] = Header(default=None)) -> None:
    expected = os.getenv("API_BEARER_TOKEN")
    if not expected:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="API_BEARER_TOKEN is not configured",
        )
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized")
    token = authorization.split(" ", 1)[1]
    if token != expected:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized")


class SearchRequest(BaseModel):
    query: str = Field(..., min_length=1)
    region: str = Field(default_factory=lambda: env_str("DDGS_REGION", "us-en") or "us-en")
    safesearch: str = Field(
        default_factory=lambda: env_str("DDGS_SAFESEARCH", "moderate") or "moderate"
    )
    timelimit: Optional[str] = Field(default_factory=lambda: env_str("DDGS_TIMELIMIT", None))
    max_results: int = Field(default_factory=lambda: env_int("DDGS_MAX_RESULTS", 10) or 10, ge=1)
    backend: Optional[str] = Field(default_factory=lambda: env_str("DDGS_BACKEND", "auto"))
    proxy: Optional[str] = Field(default_factory=lambda: env_str("DDGS_PROXY", None))
    timeout: int = Field(default_factory=lambda: env_int("DDGS_TIMEOUT", 30) or 30, ge=1)
    verify: bool = Field(default_factory=lambda: env_bool("DDGS_VERIFY", True))


def create_app(*, project_root: Path | None = None) -> FastAPI:
    load_env(project_root=project_root)
    ensure_ca_bundle(project_root=project_root)

    app = FastAPI(title="DDGS Search API", version="0.1.0")

    ui_defaults = {
        "region": env_str("DDGS_REGION", "us-en") or "us-en",
        "safesearch": env_str("DDGS_SAFESEARCH", "moderate") or "moderate",
        "timelimit": env_str("DDGS_TIMELIMIT", None),
        "max_results": env_int("DDGS_MAX_RESULTS", 10) or 10,
        "backend": env_str("DDGS_BACKEND", "auto") or "auto",
        "timeout": env_int("DDGS_TIMEOUT", 30) or 30,
        "verify": env_bool("DDGS_VERIFY", True),
    }

    @app.get("/health")
    def health() -> dict[str, str]:
        return {"status": "ok"}

    @app.get("/", response_class=HTMLResponse)
    def root() -> str:
        return render_homepage(ui_defaults)

    @app.post("/search")
    async def search(request: SearchRequest, _: None = Depends(require_auth)) -> dict[str, object]:
        results = ddgs_search(
            request.query,
            region=request.region,
            safesearch=request.safesearch,
            timelimit=request.timelimit,
            max_results=request.max_results,
            backend=request.backend,
            proxy=request.proxy,
            timeout=request.timeout,
            verify=request.verify,
        )
        return {"query": request.query, "count": len(results), "results": results}

    return app


app = create_app()