def create_app()

in scripts/e2b_router.py [0:0]

• 35 lines of code
• 5 McCabe index (conditional complexity)

def create_app(args):
    """
    Creates and configures a FastAPI application instance.
    Args:
        args: An object containing configuration parameters for the application.
              - num_sandboxes (int): The maximum number of concurrent sandboxes allowed.
    Returns:
        FastAPI: A configured FastAPI application instance.
    The application includes the following endpoints:
        1. GET /health:
            - Returns the health status of the application.
            - Response: {"status": "ok"}
        2. POST /execute_batch:
            - Executes a batch of scripts in an isolated sandbox environment.
            - Request Body: BatchRequest object containing:
                - languages (list[str]): The programming languages of the scripts (python or javascript).
                - timeout (int): The maximum execution time for each script.
                - request_timeout (int): The timeout for the request itself.
                - scripts (List[str]): A list of scripts to execute.
            - Response: A list of ScriptResult objects for each script, containing:
                - execution: The result of the script execution.
                - exception_str: Any exception encountered during execution.
    Notes:
        - A semaphore is used to limit the number of concurrent sandboxes.
        - Each script execution is wrapped in a timeout to prevent hanging.
        - Sandboxes are cleaned up after execution, even in case of errors.
    """
    app = FastAPI()

    # Instantiate semaphore and attach it to app state
    app.state.sandbox_semaphore = asyncio.Semaphore(args.max_num_sandboxes)

    @app.get("/health")
    async def health():
        return {"status": "ok"}

    @app.post("/execute_batch")
    async def execute_batch(batch: BatchRequest, request: Request):
        semaphore = request.app.state.sandbox_semaphore
        languages = batch.languages
        timeout = batch.timeout
        request_timeout = batch.request_timeout
        asyncio_timeout = batch.timeout + 1
        
        async def run_script(script: str, language: str) -> ScriptResult:

            async with semaphore:
                try:
                    sandbox = await AsyncSandbox.create(
                        timeout=timeout,
                        request_timeout=request_timeout,
                    )
                    execution = await asyncio.wait_for(
                        sandbox.run_code(script, language=language),
                        timeout=asyncio_timeout,
                    )
                    return ScriptResult(execution=execution, exception_str=None)

                except Exception as e:
                    return ScriptResult(execution=None, exception_str=str(e))
                
                finally:
                    try:
                        await sandbox.kill()
                    except Exception:
                        pass

        tasks = [run_script(script, lang) for script, lang in zip(batch.scripts, batch.languages)]
        return await asyncio.gather(*tasks)

    return app