Rewrite FastAPI app and pipeline worker entrypoint.
Worker still refuses to import backend.main. Auth router is mounted unchanged; CORS stays outermost so 401s keep headers.
This commit is contained in:
@@ -0,0 +1,178 @@
|
||||
"""FastAPI process for Periscope: lifespan, auth gate, CORS, routers.
|
||||
|
||||
Does not implement login. ``auth.router`` is mounted as-is. GCS storage is
|
||||
imported only when ``settings.use_gcs`` is true.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import JSONResponse
|
||||
from starlette.middleware.base import BaseHTTPMiddleware
|
||||
|
||||
from backend.config import settings
|
||||
from backend.routers import (
|
||||
admin,
|
||||
auth,
|
||||
contact,
|
||||
feedback,
|
||||
impedance,
|
||||
pipeline,
|
||||
projects,
|
||||
reports,
|
||||
survey,
|
||||
)
|
||||
from backend.services.projects import ProjectNotFound
|
||||
from backend.services.storage import LocalStorageBackend
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
LOCAL_DEV_USER = "local"
|
||||
|
||||
_PUBLIC_PATHS = frozenset(
|
||||
{
|
||||
"/api/contact",
|
||||
"/api/auth/mode",
|
||||
"/api/auth/register",
|
||||
"/api/auth/login",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _storage():
|
||||
if settings.use_gcs:
|
||||
from backend.services.storage_gcs import GCSStorageBackend
|
||||
|
||||
return GCSStorageBackend(settings.gcs_bucket)
|
||||
return LocalStorageBackend(settings.data_dir)
|
||||
|
||||
|
||||
def _ensure_local_dirs() -> None:
|
||||
base = settings.data_dir
|
||||
for rel in (
|
||||
"users",
|
||||
"auth/users",
|
||||
"auth/by_email",
|
||||
"library/extracted",
|
||||
"library/patterns",
|
||||
"library/models",
|
||||
"library/passives",
|
||||
"library/datasheets/refs",
|
||||
"library/datasheets/blobs",
|
||||
):
|
||||
(base / rel).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
env = os.getenv("ENVIRONMENT", "").lower()
|
||||
if env == "production" and not settings.use_auth:
|
||||
raise RuntimeError(
|
||||
"Production requires authentication: set AUTH_JWT_SECRET "
|
||||
"(local Periscope accounts) or CLERK_JWKS_URL + CLERK_SECRET_KEY."
|
||||
)
|
||||
if not settings.use_auth:
|
||||
logger.warning(
|
||||
"Authentication is DISABLED — all users have full access. "
|
||||
"This is only safe for local development."
|
||||
)
|
||||
elif settings.use_local_auth:
|
||||
logger.info("Local Periscope authentication enabled (AUTH_JWT_SECRET)")
|
||||
elif settings.use_clerk:
|
||||
logger.info("Clerk authentication enabled")
|
||||
if not settings.billing_enabled:
|
||||
logger.warning(
|
||||
"Billing is DISABLED — pipelines run free and the billing/credits "
|
||||
"routes are not mounted."
|
||||
)
|
||||
|
||||
app.state.storage = _storage()
|
||||
if isinstance(app.state.storage, LocalStorageBackend):
|
||||
_ensure_local_dirs()
|
||||
yield
|
||||
|
||||
|
||||
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
|
||||
async def dispatch(self, request: Request, call_next):
|
||||
response = await call_next(request)
|
||||
response.headers["X-Content-Type-Options"] = "nosniff"
|
||||
response.headers["X-Frame-Options"] = "DENY"
|
||||
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
|
||||
if settings.use_auth:
|
||||
response.headers["Strict-Transport-Security"] = (
|
||||
"max-age=31536000; includeSubDomains"
|
||||
)
|
||||
return response
|
||||
|
||||
|
||||
class AuthMiddleware(BaseHTTPMiddleware):
|
||||
"""Attach ``request.state.user_id``. JWT verification stays in leftover middleware."""
|
||||
|
||||
async def dispatch(self, request: Request, call_next):
|
||||
if request.method == "OPTIONS":
|
||||
return await call_next(request)
|
||||
if request.url.path in _PUBLIC_PATHS:
|
||||
request.state.user_id = LOCAL_DEV_USER
|
||||
return await call_next(request)
|
||||
|
||||
if not settings.use_auth:
|
||||
request.state.user_id = LOCAL_DEV_USER
|
||||
return await call_next(request)
|
||||
|
||||
from backend.middleware.auth import verify_request_user
|
||||
|
||||
user_id = await verify_request_user(request)
|
||||
if user_id is None:
|
||||
production = os.getenv("ENVIRONMENT", "").lower() == "production"
|
||||
if production or settings.use_local_auth:
|
||||
return JSONResponse(
|
||||
status_code=401,
|
||||
content={"detail": "Authentication required"},
|
||||
)
|
||||
user_id = LOCAL_DEV_USER
|
||||
request.state.user_id = user_id
|
||||
return await call_next(request)
|
||||
|
||||
|
||||
app = FastAPI(
|
||||
title="PeriscopeX",
|
||||
description="Agentic schematic validation API",
|
||||
lifespan=lifespan,
|
||||
)
|
||||
|
||||
# Starlette LIFO: last added is outermost. CORS must wrap Auth 401s.
|
||||
app.add_middleware(AuthMiddleware)
|
||||
app.add_middleware(SecurityHeadersMiddleware)
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=settings.cors_origins,
|
||||
allow_credentials=True,
|
||||
allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
|
||||
allow_headers=["content-type", "authorization"],
|
||||
expose_headers=["X-Datasheet-Url", "X-Datasheet-Source"],
|
||||
)
|
||||
|
||||
|
||||
@app.exception_handler(ProjectNotFound)
|
||||
async def _project_not_found_handler(request: Request, exc: ProjectNotFound):
|
||||
return JSONResponse(status_code=404, content={"detail": str(exc)})
|
||||
|
||||
|
||||
app.include_router(projects.router, prefix="/api")
|
||||
app.include_router(pipeline.router, prefix="/api")
|
||||
app.include_router(reports.router, prefix="/api")
|
||||
app.include_router(impedance.router, prefix="/api")
|
||||
app.include_router(admin.router, prefix="/api")
|
||||
app.include_router(auth.router, prefix="/api")
|
||||
if settings.billing_enabled:
|
||||
from backend.routers import billing, credits
|
||||
|
||||
app.include_router(billing.router, prefix="/api")
|
||||
app.include_router(credits.router, prefix="/api")
|
||||
app.include_router(contact.router, prefix="/api")
|
||||
app.include_router(feedback.router, prefix="/api")
|
||||
app.include_router(survey.router, prefix="/api")
|
||||
@@ -0,0 +1,109 @@
|
||||
"""Child process that runs one pipeline job. Must not import ``backend.main``."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
|
||||
from backend.config import settings
|
||||
from backend.services import event_bridge
|
||||
from backend.services import pipeline as pipeline_svc
|
||||
from backend.services.storage import LocalStorageBackend, StorageBackend
|
||||
|
||||
_TRUE = frozenset({"1", "true", "yes", "on"})
|
||||
|
||||
|
||||
def _storage() -> StorageBackend:
|
||||
if settings.use_gcs:
|
||||
from backend.services.storage_gcs import GCSStorageBackend
|
||||
|
||||
return GCSStorageBackend(settings.gcs_bucket)
|
||||
return LocalStorageBackend(settings.data_dir)
|
||||
|
||||
|
||||
def _need(name: str) -> str:
|
||||
value = os.environ.get(name, "").strip()
|
||||
if not value:
|
||||
raise SystemExit(f"missing required env var: {name}")
|
||||
return value
|
||||
|
||||
|
||||
def _flag(name: str, default: bool = False) -> bool:
|
||||
raw = os.environ.get(name, "").strip().lower()
|
||||
if not raw:
|
||||
return default
|
||||
return raw in _TRUE
|
||||
|
||||
|
||||
async def _dispatch() -> None:
|
||||
project_id = _need("PROJECT_ID")
|
||||
user_id = _need("USER_ID")
|
||||
resume = _flag("RESUME")
|
||||
free = _flag("FREE")
|
||||
mode = (os.environ.get("MODE", "run").strip().lower() or "run")
|
||||
execution_name = os.environ.get("EXECUTION_NAME", "").strip()
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [worker %(name)s] %(message)s",
|
||||
)
|
||||
log = logging.getLogger("backend.pipeline_worker")
|
||||
log.info(
|
||||
"starting worker mode=%s project=%s user=%s resume=%s free=%s execution=%s",
|
||||
mode,
|
||||
project_id,
|
||||
user_id,
|
||||
resume,
|
||||
free,
|
||||
execution_name or "(none)",
|
||||
)
|
||||
|
||||
storage = _storage()
|
||||
pipeline_svc.set_broker(event_bridge.GCSEventBroker(storage, user_id))
|
||||
pipeline_svc.broker.clear_history(project_id)
|
||||
|
||||
if mode == "run":
|
||||
await pipeline_svc.run_pipeline(
|
||||
storage, user_id, project_id, resume=resume, free=free
|
||||
)
|
||||
return
|
||||
if mode == "regen":
|
||||
stages = [
|
||||
part.strip()
|
||||
for part in os.environ.get("REGEN_STAGES", "").split(",")
|
||||
if part.strip()
|
||||
]
|
||||
if not stages:
|
||||
raise SystemExit("REGEN_STAGES must list at least one stage in regen mode")
|
||||
await pipeline_svc.run_regen_pipeline(storage, user_id, project_id, stages)
|
||||
return
|
||||
if mode == "placement":
|
||||
from backend.services import placement_pipeline
|
||||
|
||||
await placement_pipeline.run_placement_pipeline(storage, user_id, project_id)
|
||||
return
|
||||
if mode == "pcb":
|
||||
from backend.services import pcb_pipeline
|
||||
|
||||
await pcb_pipeline.run_pcb_pipeline(storage, user_id, project_id)
|
||||
return
|
||||
raise SystemExit(
|
||||
f"unknown MODE={mode!r}; expected 'run', 'regen', 'placement', or 'pcb'"
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
try:
|
||||
asyncio.run(_dispatch())
|
||||
except SystemExit:
|
||||
raise
|
||||
except KeyboardInterrupt:
|
||||
sys.exit(130)
|
||||
except BaseException as exc: # pragma: no cover
|
||||
logging.exception("worker crashed before run_pipeline cleanup: %s", exc)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,45 @@
|
||||
"""Native FastAPI app and worker entrypoint resolve from src."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
import backend.main as main_mod
|
||||
import backend.pipeline_worker as worker_mod
|
||||
from backend.main import LOCAL_DEV_USER, app
|
||||
|
||||
|
||||
def _src(mod, name: str) -> Path:
|
||||
path = Path(mod.__file__).resolve()
|
||||
assert path.name == name
|
||||
assert "src" in path.parts
|
||||
assert "Native Periscope overlay" not in path.read_text(encoding="utf-8")[:400]
|
||||
return path
|
||||
|
||||
|
||||
def test_main_and_worker_are_src():
|
||||
_src(main_mod, "main.py")
|
||||
_src(worker_mod, "pipeline_worker.py")
|
||||
|
||||
|
||||
def test_worker_does_not_import_fastapi_app():
|
||||
tree = ast.parse(Path(worker_mod.__file__).read_text(encoding="utf-8"))
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Import):
|
||||
for alias in node.names:
|
||||
assert alias.name != "backend.main"
|
||||
if isinstance(node, ast.ImportFrom) and node.module:
|
||||
assert not node.module.startswith("backend.main")
|
||||
if node.module == "backend":
|
||||
for alias in node.names:
|
||||
assert alias.name != "main"
|
||||
|
||||
|
||||
def test_app_mounts_auth_and_public_user():
|
||||
paths = set(app.openapi()["paths"])
|
||||
assert any(p.startswith("/api/auth") for p in paths)
|
||||
assert any(p.startswith("/api/projects") for p in paths)
|
||||
assert LOCAL_DEV_USER == "local"
|
||||
assert "/api/auth/login" in main_mod._PUBLIC_PATHS
|
||||
assert "/api/contact" in main_mod._PUBLIC_PATHS
|
||||
Reference in New Issue
Block a user