Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
tross-assignment app.py
3.1 kB · 104 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105"""FastAPI application entrypoint.
Wires the API routers, service-error exception handling, and startupconfiguration validation. Run from the project root::
uvicorn app:app
Configuration is loaded from ``config.toml`` (see :mod:`src.config`). If anyrequired setting is missing or incomplete the app refuses to start, raising``RuntimeError`` from the lifespan startup hook so the failure is loud andearly rather than surfacing on the first request."""
from collections.abc import AsyncGeneratorfrom contextlib import asynccontextmanager
from fastapi import ( FastAPI, Request, Response,)from fastapi.responses import JSONResponse
from src.api import router as api_routerfrom src.config import configfrom src.exceptions import AppServiceError
# LinkedIn session cookies that the scraper requires to authenticate._REQUIRED_COOKIES: tuple[str, ...] = ("li_at", "JSESSIONID")
def _validate_config() -> None: """Fail fast at startup if required configuration is missing or incomplete.
Raises: RuntimeError: If the admin API key is unset or any required LinkedIn session cookie is missing. """ problems: list[str] = []
if not config.admin_key: problems.append("admin_key is not set in config.toml") if not config.tross_key: problems.append("tross_key is not set in config.toml")
cookies = config.linkedin.cookies missing = [name for name in _REQUIRED_COOKIES if not cookies.get(name)] if missing: problems.append(f"missing LinkedIn cookies: {', '.join(missing)}")
if problems: raise RuntimeError("application configuration incomplete: " + "; ".join(problems))
@asynccontextmanagerasync def lifespan(app: FastAPI) -> AsyncGenerator[None]: """Validate configuration before serving requests, then run the app.""" _validate_config() yield
app = FastAPI(title="Tross LinkedIn Scraper", lifespan=lifespan)app.include_router(api_router)
@app.get("/health", tags=["Health"])async def health() -> dict[str, str]: """Liveness probe (no auth).""" return {"status": "ok"}
@app.get("/", include_in_schema=False)def hello() -> Response: ret = r""" ________________________________________/ This is the most efficient endpoint in \\ this project! / ---------------------------------------- \ ^__^ \ (oo)\_______ (__)\ )\/\ ||----w | || ||
""" return Response(status_code=200, content=ret)
@app.exception_handler(AppServiceError)async def app_service_error_handler(request: Request, exc: AppServiceError) -> JSONResponse: """Map service-layer errors to structured JSON responses.
Args: request: The incoming request that triggered the error. exc: The raised :class:`AppServiceError`.
Returns: A JSON response carrying the error code and message at the error's configured status code. """ return JSONResponse( status_code=exc.status_code, content={"error_code": exc.error_code, "message": exc.message}, )