app/api/v1/code_executor.py

from __future__ import annotations

import time
from typing import Optional

from fastapi import APIRouter, Depends

from app.api.deps import require_auth
from app.core.logger import get_logger
from app.models.schemas import (
    CodeExecutionItemResult,
    CodeExecutionRequest,
    CodeExecutionResponse,
)
from app.services.code_executor_service import CodeExecutorService

router = APIRouter()
_logger = get_logger(__name__)

_executor: Optional[CodeExecutorService] = None


def get_executor() -> CodeExecutorService:
    global _executor
    if _executor is None:
        _executor = CodeExecutorService()
    return _executor


@router.post(
    "/code/execute",
    response_model=CodeExecutionResponse,
    summary="Execute code snippets (Python, JavaScript, Java)",
)
async def execute_code(
    body: CodeExecutionRequest,
    token: str = Depends(require_auth),
    executor: CodeExecutorService = Depends(get_executor),
) -> CodeExecutionResponse:
    _logger.info("Code execution request: items=%s", len(body.items))

    start = time.perf_counter()
    results: list[CodeExecutionItemResult] = []

    for item in body.items:
        t0 = time.perf_counter()
        try:
            result = await executor.execute(item.code, item.language)
            elapsed = (time.perf_counter() - t0) * 1000
            results.append(CodeExecutionItemResult(
                success=result["success"],
                output=result.get("output", ""),
                error=result.get("error"),
                exit_code=result.get("exit_code"),
                execution_time_ms=round(elapsed, 2),
                language=result.get("language", item.language),
                timed_out=result.get("timed_out", False),
            ))
        except Exception as exc:
            elapsed = (time.perf_counter() - t0) * 1000
            _logger.exception("Item execution error")
            results.append(CodeExecutionItemResult(
                success=False,
                error=str(exc),
                execution_time_ms=round(elapsed, 2),
                language=item.language,
            ))

    total_ms = (time.perf_counter() - start) * 1000
    success_count = sum(1 for r in results if r.success)
    failed_count = len(results) - success_count
    all_ok = failed_count == 0

    _logger.info(
        "Code execution done: items=%s, success=%s, failed=%s, total_ms=%s",
        len(results), success_count, failed_count, round(total_ms, 3),
    )
    return CodeExecutionResponse(
        success=all_ok,
        time_ms=round(total_ms, 3),
        success_count=success_count,
        failed_count=failed_count,
        results=results,
    )