Spaces:
Sleeping
Sleeping
File size: 3,790 Bytes
96d51f5 | 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 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | """FastAPI application for the Python code review environment."""
from __future__ import annotations
import os
from fastapi import APIRouter, HTTPException
from fastapi.responses import RedirectResponse
try:
from compat import create_app
from models import (
HealthResponse,
PythonCodeReviewAction,
PythonCodeReviewObservation,
PythonCodeReviewState,
TaskDescriptor,
TaskGrade,
)
except Exception:
from .compat import create_app
from .models import (
HealthResponse,
PythonCodeReviewAction,
PythonCodeReviewObservation,
PythonCodeReviewState,
TaskDescriptor,
TaskGrade,
)
from server.env import PythonCodeReviewEnvironment
try:
MAX_CONCURRENT_ENVS = max(int(os.getenv("MAX_CONCURRENT_ENVS", "16")), 1)
except Exception:
MAX_CONCURRENT_ENVS = 16
python_env = PythonCodeReviewEnvironment(verbose=False)
app = create_app(
PythonCodeReviewEnvironment,
PythonCodeReviewAction,
PythonCodeReviewObservation,
max_concurrent_envs=MAX_CONCURRENT_ENVS,
)
router = APIRouter(tags=["python-code-review"])
@router.get("/", include_in_schema=False)
def root() -> RedirectResponse:
"""Redirect root to API documentation."""
return RedirectResponse(url="/docs")
@router.get("/health", response_model=HealthResponse)
def health() -> HealthResponse:
"""Health check endpoint for deployment monitoring."""
return python_env.health()
@router.get("/tasks", response_model=list)
def list_tasks() -> list:
"""List all available deterministic tasks."""
return python_env.list_task_summaries()
@router.get("/tasks/{task_id}", response_model=object)
def get_task(task_id: str) -> object:
"""Get a specific task by ID."""
try:
return python_env.get_task(task_id)
except ValueError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@router.post("/tasks/{task_id}/grade", response_model=TaskGrade)
def grade_task(task_id: str, payload: PythonCodeReviewAction) -> TaskGrade:
"""Grade code submission for a task without running an episode."""
if payload.action_type != "edit_code" or not payload.code:
raise HTTPException(
status_code=400,
detail="Requires action_type='edit_code' with code parameter."
)
try:
return python_env.grade_task_submission(task_id=task_id, code=payload.code)
except ValueError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@router.post("/state", response_model=PythonCodeReviewState)
def get_state_post() -> RedirectResponse:
"""Redirect POST /state to GET for compatibility."""
return RedirectResponse(url="/state", status_code=303)
app.include_router(router)
def _prioritize_route(path: str, methods: set[str]) -> None:
"""Move a matching custom route ahead of default OpenEnv routes."""
try:
for index in range(len(app.router.routes) - 1, -1, -1):
route = app.router.routes[index]
route_path = getattr(route, "path", None)
route_methods = set(getattr(route, "methods", set()) or set())
if route_path == path and methods.issubset(route_methods):
app.router.routes.insert(0, app.router.routes.pop(index))
break
except Exception:
pass
_prioritize_route("/health", {"GET"})
def main(host: str = "0.0.0.0", port: int = 8000) -> None:
"""Run the FastAPI application with uvicorn."""
import uvicorn
uvicorn.run(
app,
host=os.getenv("HOST", host),
port=int(os.getenv("PORT", str(port))),
)
if __name__ == "__main__":
main()
|