File size: 3,324 Bytes
2aa3e66 1cb0a2e 2aa3e66 1cb0a2e 2aa3e66 1cb0a2e 2aa3e66 1cb0a2e 2aa3e66 1cb0a2e 2aa3e66 | 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 | 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()
|