API Reference

def run(
    func_or_list: Callable[..., Any] | list[Callable[..., Any]] | dict[str, list[Callable[..., Any]]], 
    host: str = "0.0.0.0", 
    port: int = 8000, 
    auth: dict[str, str] | None = None,
    secret_key: str | None = None,
    uploads_dir: str | Path = "./uploads",
    returns_dir: str | Path = "./returned_files",
    auto_delete_uploads: bool = True,
    template_dir: str | Path | None = None,
    root_path: str = "",
    fastapi_config: dict[str, Any] | None = None,
    **kwargs
) -> None:
    """Generate and run a web UI for one or more Python functions.

    Single function mode: Creates a form at root (/) for the function.
    Multiple functions mode: Creates an index page with links to individual function forms.
    Grouped functions mode: Creates an index page with grouped sections of functions.

    Args:
        func_or_list: A single function, list of functions, or dict of {group_name: [functions]}.
        host: Server host address (default: "0.0.0.0").
        port: Server port (default: 8000).
        auth: Optional dictionary of {username: password} for authentication.
        secret_key: Secret key for session signing (required if auth is used). 
                    If None, a random one is generated on startup.
        uploads_dir: Directory for uploaded files (default: "./uploads").
        returns_dir: Directory for returned files (default: "./returned_files").
        auto_delete_uploads: If True, delete uploaded files after processing (default: True).
        template_dir: Optional custom template directory.
        root_path: Prefix for the API path (useful for reverse proxies).
        fastapi_config: Optional dictionary with extra arguments for FastAPI app 
                        (e.g. {'title': 'My App', 'version': '1.0.0'}).
        **kwargs: Extra options passed directly to `uvicorn.Config`.
                  Examples: `ssl_keyfile`, `ssl_certfile`, `log_level`, `workers`.

    Raises:
        FileNotFoundError: If template directory doesn't exist.
        TypeError: If function parameters use unsupported types.

    Examples:
        Single function:
            run(my_function)

        Multiple functions:
            run([func1, func2, func3])

        Grouped functions:
            run({
                'Math': [add, subtract, multiply],
                'Text': [uppercase, lowercase]
            })

    Notes:
        - Returned files are automatically deleted 1 hour after creation (hardcoded).
        - Cleanup runs on startup and then every hour while server is running.
        - Multiple workers are supported (each worker runs its own cleanup task).
    """

    uploads_path = Path(uploads_dir)
    returns_path = Path(returns_dir)
    uploads_path.mkdir(parents=True, exist_ok=True)
    returns_path.mkdir(parents=True, exist_ok=True)

    config.UPLOADS_DIR = uploads_path
    config.RETURNS_DIR = returns_path
    config.AUTO_DELETE_UPLOADS = auto_delete_uploads

    is_grouped = isinstance(func_or_list, dict)
    is_single = not isinstance(func_or_list, (list, dict))

    if is_grouped:
        grouped_funcs = func_or_list
        funcs = []
        for group_functions in grouped_funcs.values():
            funcs.extend(group_functions)
    elif is_single:
        funcs = [func_or_list]
    else:
        funcs = func_or_list

    app_kwargs = {"root_path": root_path}

    if fastapi_config:
        conf = fastapi_config.copy()
        if "root_path" in conf:
            conf.pop("root_path") 
        app_kwargs.update(conf)

    app = FastAPI(**app_kwargs)

    @app.on_event("startup")
    async def startup_cleanup():
        """Cleanup old files on startup and run periodic cleanup task."""
        await asyncio.to_thread(cleanup_old_files)

        file_count = get_returned_files_count()
        if file_count > 10000:
            warnings.warn(
                f"Returns directory has {file_count} files. "
                "Consider manually cleaning old files or restarting the server.",
                UserWarning
            )

        async def periodic_cleanup_task():
            """Run cleanup every hour to remove files older than 1 hour."""
            while True:
                await asyncio.sleep(3600)
                await asyncio.to_thread(cleanup_old_files)

        asyncio.create_task(periodic_cleanup_task())

    if template_dir is None:
        template_dir = Path(__file__).parent / "templates"
    else:
        template_dir = Path(template_dir)

    if not template_dir.exists():
        raise FileNotFoundError(f"Template directory '{template_dir}' not found.")

    templates = Jinja2Templates(directory=str(template_dir))
    app.mount("/static", StaticFiles(directory=template_dir / "static"), name="static")

    setup_download_route(app)

    if is_single:
        func = funcs[0]
        params = analyze(func)
        setup_single_function_routes(app, func, params, templates, bool(auth))
    elif is_grouped:
        setup_grouped_function_routes(app, grouped_funcs, templates, bool(auth))
    else:
        setup_multiple_function_routes(app, funcs, templates, bool(auth))

    if auth:
        setup_auth_middleware(app, auth, templates, secret_key)

    uvicorn_params = {
        "host": host,
        "port": port,
        "reload": False,
        "limit_concurrency": 100,
        "limit_max_requests": 1000,
        "timeout_keep_alive": 30,
        "h11_max_incomplete_event_size": 16 * 1024 * 1024
    }

    if "root_path" in kwargs:
        kwargs.pop("root_path")

    uvicorn_params.update(kwargs)

    config_obj = uvicorn.Config(app, **uvicorn_params)
    server = uvicorn.Server(config_obj)
    asyncio.run(server.serve())

def run(
    func_or_list: Callable[..., Any] | list[Callable[..., Any]] | dict[str, list[Callable[..., Any]]], 
    host: str = "0.0.0.0", 
    port: int = 8000, 
    auth: dict[str, str] | None = None,
    secret_key: str | None = None,
    uploads_dir: str | Path = "./uploads",
    returns_dir: str | Path = "./returned_files",
    auto_delete_uploads: bool = True,
    template_dir: str | Path | None = None,
    root_path: str = "",
    fastapi_config: dict[str, Any] | None = None,
    **kwargs
) -> None:
    """Generate and run a web UI for one or more Python functions.

    Single function mode: Creates a form at root (/) for the function.
    Multiple functions mode: Creates an index page with links to individual function forms.
    Grouped functions mode: Creates an index page with grouped sections of functions.

    Args:
        func_or_list: A single function, list of functions, or dict of {group_name: [functions]}.
        host: Server host address (default: "0.0.0.0").
        port: Server port (default: 8000).
        auth: Optional dictionary of {username: password} for authentication.
        secret_key: Secret key for session signing (required if auth is used). 
                    If None, a random one is generated on startup.
        uploads_dir: Directory for uploaded files (default: "./uploads").
        returns_dir: Directory for returned files (default: "./returned_files").
        auto_delete_uploads: If True, delete uploaded files after processing (default: True).
        template_dir: Optional custom template directory.
        root_path: Prefix for the API path (useful for reverse proxies).
        fastapi_config: Optional dictionary with extra arguments for FastAPI app 
                        (e.g. {'title': 'My App', 'version': '1.0.0'}).
        **kwargs: Extra options passed directly to `uvicorn.Config`.
                  Examples: `ssl_keyfile`, `ssl_certfile`, `log_level`, `workers`.

    Raises:
        FileNotFoundError: If template directory doesn't exist.
        TypeError: If function parameters use unsupported types.

    Examples:
        Single function:
            run(my_function)

        Multiple functions:
            run([func1, func2, func3])

        Grouped functions:
            run({
                'Math': [add, subtract, multiply],
                'Text': [uppercase, lowercase]
            })

    Notes:
        - Returned files are automatically deleted 1 hour after creation (hardcoded).
        - Cleanup runs on startup and then every hour while server is running.
        - Multiple workers are supported (each worker runs its own cleanup task).
    """

    uploads_path = Path(uploads_dir)
    returns_path = Path(returns_dir)
    uploads_path.mkdir(parents=True, exist_ok=True)
    returns_path.mkdir(parents=True, exist_ok=True)

    config.UPLOADS_DIR = uploads_path
    config.RETURNS_DIR = returns_path
    config.AUTO_DELETE_UPLOADS = auto_delete_uploads

    is_grouped = isinstance(func_or_list, dict)
    is_single = not isinstance(func_or_list, (list, dict))

    if is_grouped:
        grouped_funcs = func_or_list
        funcs = []
        for group_functions in grouped_funcs.values():
            funcs.extend(group_functions)
    elif is_single:
        funcs = [func_or_list]
    else:
        funcs = func_or_list

    app_kwargs = {"root_path": root_path}

    if fastapi_config:
        conf = fastapi_config.copy()
        if "root_path" in conf:
            conf.pop("root_path") 
        app_kwargs.update(conf)

    app = FastAPI(**app_kwargs)

    @app.on_event("startup")
    async def startup_cleanup():
        """Cleanup old files on startup and run periodic cleanup task."""
        await asyncio.to_thread(cleanup_old_files)

        file_count = get_returned_files_count()
        if file_count > 10000:
            warnings.warn(
                f"Returns directory has {file_count} files. "
                "Consider manually cleaning old files or restarting the server.",
                UserWarning
            )

        async def periodic_cleanup_task():
            """Run cleanup every hour to remove files older than 1 hour."""
            while True:
                await asyncio.sleep(3600)
                await asyncio.to_thread(cleanup_old_files)

        asyncio.create_task(periodic_cleanup_task())

    if template_dir is None:
        template_dir = Path(__file__).parent / "templates"
    else:
        template_dir = Path(template_dir)

    if not template_dir.exists():
        raise FileNotFoundError(f"Template directory '{template_dir}' not found.")

    templates = Jinja2Templates(directory=str(template_dir))
    app.mount("/static", StaticFiles(directory=template_dir / "static"), name="static")

    setup_download_route(app)

    if is_single:
        func = funcs[0]
        params = analyze(func)
        setup_single_function_routes(app, func, params, templates, bool(auth))
    elif is_grouped:
        setup_grouped_function_routes(app, grouped_funcs, templates, bool(auth))
    else:
        setup_multiple_function_routes(app, funcs, templates, bool(auth))

    if auth:
        setup_auth_middleware(app, auth, templates, secret_key)

    uvicorn_params = {
        "host": host,
        "port": port,
        "reload": False,
        "limit_concurrency": 100,
        "limit_max_requests": 1000,
        "timeout_keep_alive": 30,
        "h11_max_incomplete_event_size": 16 * 1024 * 1024
    }

    if "root_path" in kwargs:
        kwargs.pop("root_path")

    uvicorn_params.update(kwargs)

    config_obj = uvicorn.Config(app, **uvicorn_params)
    server = uvicorn.Server(config_obj)
    asyncio.run(server.serve())

def cleanup_old_files() -> None:
    """Remove RETURNED files older than 1 hour (hardcoded).

    This runs on startup and every hour while server is running.
    Only affects files in the returns directory (FileResponse outputs).
    Uploaded files are cleaned up immediately after processing if AUTO_DELETE_UPLOADS is True.
    """
    try:
        max_age_seconds = config.RETURNS_LIFETIME_SECONDS
        old_file_ids = get_old_returned_files(max_age_seconds)

        for file_id in old_file_ids:
            cleanup_returned_file(file_id, delete_from_disk=True)

    except Exception:
        pass

def cleanup_returned_file(file_id: str, delete_from_disk: bool = True) -> None:
    """Remove RETURNED file from disk.

    Args:
        file_id: Unique identifier for the returned file.
        delete_from_disk: If True, delete the physical file from disk.
    """
    try:
        if delete_from_disk:
            for file_path in config.RETURNS_DIR.iterdir():
                if file_path.is_file():
                    metadata = _decode_filename(file_path.name)
                    if metadata and metadata['file_id'] == file_id:
                        try:
                            os.unlink(file_path)
                        except FileNotFoundError:
                            pass
                        break
    except Exception:
        pass

def cleanup_uploaded_file(file_path: str) -> None:
    """Delete an UPLOADED file from disk.

    Args:
        file_path: Path to the uploaded file.
    """
    try:
        if os.path.exists(file_path):
            os.unlink(file_path)
    except Exception:
        pass

def create_response_with_files(processed: dict[str, Any]) -> dict[str, Any]:
    """Create JSON response with RETURNED file downloads.

    Converts file paths to file_ids for the download endpoint.
    Only processes files returned by user functions (FileResponse).

    Args:
        processed: Processed result from process_result().

    Returns:
        Response dictionary with file IDs and metadata.
    """
    response = {"success": True, "result_type": processed['type']}

    if processed['type'] == 'download':
        path = processed['path']
        filename = Path(path).name
        metadata = _decode_filename(filename)

        if metadata:
            response['file_id'] = metadata['file_id']
            response['filename'] = processed['filename']
        else:
            response['file_id'] = 'unknown'
            response['filename'] = processed['filename']

    elif processed['type'] == 'downloads':
        files = []
        for f in processed['files']:
            path = f['path']
            filename_on_disk = Path(path).name
            metadata = _decode_filename(filename_on_disk)

            if metadata:
                files.append({
                    'file_id': metadata['file_id'],
                    'filename': f['filename']
                })
            else:
                files.append({
                    'file_id': 'unknown',
                    'filename': f['filename']
                })
        response['files'] = files

    elif processed['type'] == 'multiple':
        outputs = []
        for output in processed['outputs']:
            output_response = create_response_with_files(output)
            output_response.pop('success', None)
            outputs.append(output_response)
        response['outputs'] = outputs

    elif processed['type'] == 'table':
        response['headers'] = processed['headers']
        response['rows'] = processed['rows']

    else:
        response['result'] = processed['data']

    return response

def get_old_returned_files(max_age_seconds: int) -> list[str]:
    """Get file_ids of RETURNED files older than max_age_seconds.

    Parses timestamps from filenames and compares against current time.

    Args:
        max_age_seconds: Maximum age in seconds.

    Returns:
        List of file_ids (strings) for returned files.
    """
    try:
        current_time = int(time.time())
        old_file_ids = []

        for file_path in config.RETURNS_DIR.iterdir():
            if file_path.is_file():
                metadata = _decode_filename(file_path.name)
                if metadata:
                    age = current_time - metadata['timestamp']
                    if age > max_age_seconds:
                        old_file_ids.append(metadata['file_id'])

        return old_file_ids
    except Exception:
        return []

def get_returned_file(file_id: str) -> dict[str, str] | None:
    """Get RETURNED file info by file_id.

    Searches the returns directory for a file matching the file_id.

    Args:
        file_id: Unique identifier for the returned file.

    Returns:
        Dictionary with 'path' and 'filename' keys, or None if not found.
    """
    try:
        for file_path in config.RETURNS_DIR.iterdir():
            if file_path.is_file():
                metadata = _decode_filename(file_path.name)
                if metadata and metadata['file_id'] == file_id:
                    return {
                        'path': str(file_path),
                        'filename': metadata['filename']
                    }
        return None
    except Exception:
        return None

def get_returned_files_count() -> int:
    """Get count of RETURNED files in directory.

    Returns:
        Number of valid returned files.
    """
    try:
        count = 0
        for file_path in config.RETURNS_DIR.iterdir():
            if file_path.is_file():
                metadata = _decode_filename(file_path.name)
                if metadata:
                    count += 1
        return count
    except Exception:
        return 0

def save_returned_file(data: bytes, filename: str) -> tuple[str, str]:
    """Save a RETURNED file (from user's FileResponse) to returns directory.

    This is SAFE because WE control the filename (it comes from user's code, not from upload).
    The filename is already validated by Pydantic (max 150 chars) in FileResponse.

    Args:
        data: File content bytes.
        filename: Filename from user's FileResponse (already validated).

    Returns:
        Tuple of (file_id, file_path).
    """
    file_id = uuid.uuid4().hex
    timestamp = int(time.time())
    encoded_name = _encode_filename(file_id, timestamp, filename)
    file_path = config.RETURNS_DIR / encoded_name

    with open(file_path, 'wb') as f:
        f.write(data)

    return file_id, str(file_path)

async def save_uploaded_file(uploaded_file: Any, suffix: str) -> str:
    """Save an UPLOADED file (received from user) to uploads directory.

    Sanitizes filename, preserves original name (up to 100 chars), and adds unique ID.
    Format: {sanitized_name}_{unique_id}.{ext}
    Example: My_Report_2024_a3b4c5d6e7f8g9h0i1j2k3l4m5n6o7p8.pdf

    Security features:
    - Filename sanitization against directory traversal
    - 100 character limit on user-provided name
    - UUID for uniqueness filename
    - Path resolution check to ensure file stays in uploads directory

    Args:
        uploaded_file: The uploaded file object from FastAPI.
        suffix: File extension to use as fallback.

    Returns:
        Path to the saved file (string).

    Raises:
        ValueError: If security check fails (file would be outside uploads directory).
    """
    original_name = uploaded_file.filename if hasattr(uploaded_file, 'filename') else 'file'
    safe_name = _sanitize_filename(original_name)

    name_without_ext, ext = os.path.splitext(safe_name)
    if not ext:
        ext = suffix

    if len(name_without_ext) > MAX_USER_FILENAME_LENGTH:
        name_without_ext = name_without_ext[:MAX_USER_FILENAME_LENGTH]

    unique_id = uuid.uuid4().hex
    final_name = f"{name_without_ext}_{unique_id}{ext}"

    file_path = config.UPLOADS_DIR / final_name
    file_path_resolved = file_path.resolve()
    uploads_dir_resolved = config.UPLOADS_DIR.resolve()

    if not str(file_path_resolved).startswith(str(uploads_dir_resolved)):
        raise ValueError("Security: Invalid file path detected")

    with open(file_path, 'wb', buffering=FILE_BUFFER_SIZE) as f:
        while chunk := await uploaded_file.read(CHUNK_SIZE):
            f.write(chunk)

    return str(file_path)

def setup_auth_middleware(app, auth: dict[str, str], templates: Jinja2Templates, secret_key: str | None = None):
    """Setup authentication middleware and routes.

    Args:
        app: FastAPI application instance.
        auth: Dictionary of {username: password} for authentication.
        templates: Jinja2Templates instance.
        secret_key: Secret key for session signing.
    """
    key = secret_key or secrets.token_hex(32)

    # 1. Define Auth Middleware (INNER)
    @app.middleware("http")
    async def auth_middleware(request: Request, call_next):
        path = request.url.path

        # Allow public paths: login page, auth endpoint, and static assets
        if path in ["/login", "/auth"] or path.startswith("/static"):
            return await call_next(request)

        # Check for valid session
        if not request.session.get("user"):
            # If API call (AJAX), return 401
            if "application/json" in request.headers.get("accept", ""):
                return JSONResponse({"error": "Unauthorized"}, status_code=401)
            # If browser navigation, redirect to login
            return RedirectResponse(url="/login")

        return await call_next(request)

    # 2. Add SessionMiddleware (OUTER - runs first)
    app.add_middleware(SessionMiddleware, secret_key=key, https_only=False)

    @app.get("/login")
    async def login_page(request: Request):
        # If already logged in, go home
        if request.session.get("user"):
            return RedirectResponse(url="/")
        return templates.TemplateResponse("login.html", {"request": request})

    @app.post("/auth")
    async def authenticate(request: Request):
        try:
            form = await request.form()
            username = form.get("username")
            password = form.get("password")

            if username in auth:
                # Safe comparison against Timing Attacks
                if secrets.compare_digest(auth[username], password):
                    request.session["user"] = username
                    return RedirectResponse(url="/", status_code=303)

            return templates.TemplateResponse(
                "login.html", 
                {"request": request, "error": "Invalid credentials"}
            )
        except Exception:
            return templates.TemplateResponse(
                "login.html", 
                {"request": request, "error": "Login failed"}
            )

    @app.get("/logout")
    async def logout(request: Request):
        request.session.clear()
        return RedirectResponse(url="/login")

async def handle_form_submission(
    request: Request, 
    func: Callable, 
    params: dict[str, ParamInfo]
) -> JSONResponse:
    """Handle form submission for any function.

    Args:
        request: FastAPI request object.
        func: Function to call with validated parameters.
        params: Parameter metadata from analyze().

    Returns:
        JSON response with result or error.
    """
    uploaded_files = []

    try:
        form_data = await request.form()
        data = {}

        for name, info in params.items():
            if info.is_list:
                raw_values = form_data.getlist(name)

                if not raw_values and name in form_data:
                     raw_values = [form_data[name]]

                processed_list = []
                for val in raw_values:
                    if hasattr(val, 'filename'):
                        suffix = os.path.splitext(val.filename)[1]
                        file_path = await save_uploaded_file(val, suffix)
                        uploaded_files.append(file_path)
                        processed_list.append(file_path)
                    else:
                        processed_list.append(val)

                if len(processed_list) == 1 and isinstance(processed_list[0], str) and processed_list[0].startswith('['):
                    data[name] = processed_list[0]
                else:
                    data[name] = processed_list

            else:
                value = form_data.get(name)
                if hasattr(value, 'filename'):
                    suffix = os.path.splitext(value.filename)[1]
                    file_path = await save_uploaded_file(value, suffix)
                    uploaded_files.append(file_path)
                    data[name] = file_path
                else:
                    data[name] = value

        for key, value in form_data.items():
            if key.endswith('_optional_toggle'):
                data[key] = value

        validated = validate_params(data, params)

        if inspect.iscoroutinefunction(func):
            result = await func(**validated)
        else:
            result = await asyncio.to_thread(func, **validated)

        if config.AUTO_DELETE_UPLOADS:
            for file_path in uploaded_files:
                cleanup_uploaded_file(file_path)

        processed = await asyncio.to_thread(process_result, result)
        response = create_response_with_files(processed)

        return JSONResponse(response)

    except Exception as e:
        if config.AUTO_DELETE_UPLOADS:
            for file_path in uploaded_files:
                cleanup_uploaded_file(file_path)

        return JSONResponse({"success": False, "error": str(e)}, status_code=400)

def setup_download_route(app):
    """Setup file download route.

    Args:
        app: FastAPI application instance.
    """
    @app.get("/download/{file_id}")
    async def download_file(file_id: str):
        if not UUID_PATTERN.match(file_id):
            return JSONResponse({"error": "Invalid file ID"}, status_code=400)

        file_info = get_returned_file(file_id)

        if not file_info:
            return JSONResponse({"error": "File not found"}, status_code=404)

        path = file_info['path']
        filename = file_info['filename']

        if not os.path.exists(path):
            cleanup_returned_file(file_id, delete_from_disk=False)
            return JSONResponse({"error": "File expired"}, status_code=404)

        safe_filename = os.path.basename(filename)

        response = FastAPIFileResponse(
            path=path,
            filename=safe_filename,
            media_type='application/octet-stream'
        )

        return response

def setup_grouped_function_routes(app, grouped_funcs: dict[str, list[Callable]], templates: Jinja2Templates, has_auth: bool):
    """Setup routes for grouped functions mode.

    Args:
        app: FastAPI application instance.
        grouped_funcs: Dictionary of {group_name: [functions]}.
        templates: Jinja2Templates instance.
        has_auth: Whether authentication is enabled.
    """
    @app.get("/")
    async def index(request: Request):
        groups = []
        for group_name, funcs in grouped_funcs.items():
            tools = [{
                "name": f.__name__.replace('_', ' ').title(),
                "path": f"/{f.__name__}"
            } for f in funcs]

            groups.append({
                "name": group_name,
                "tools": tools
            })

        return templates.TemplateResponse(
            "index.html",
            {"request": request, "groups": groups, "has_auth": has_auth}
        )

    for funcs in grouped_funcs.values():
        for func in funcs:
            _register_function_routes(app, func, templates, has_auth)

def setup_multiple_function_routes(app, funcs: list[Callable], templates: Jinja2Templates, has_auth: bool):
    """Setup routes for multiple functions mode.

    Args:
        app: FastAPI application instance.
        funcs: List of functions to wrap.
        templates: Jinja2Templates instance.
        has_auth: Whether authentication is enabled.
    """
    @app.get("/")
    async def index(request: Request):
        tools = [{
            "name": f.__name__.replace('_', ' ').title(),
            "path": f"/{f.__name__}"
        } for f in funcs]
        return templates.TemplateResponse(
            "index.html",
            {"request": request, "tools": tools, "has_auth": has_auth}
        )

    for func in funcs:
        _register_function_routes(app, func, templates, has_auth)