Click CLI to FastAPI Conversion: The Machine's Reckoning¶

# factory.py - The conversion engine
from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, UploadFile, File
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import StreamingResponse, FileResponse
from pydantic import BaseModel, create_model, Field, validator
from typing import Any, Dict, List, Optional, Union, Type
import click
import inspect
import asyncio
import tempfile
import os
from pathlib import Path
import json
import io

def _click_option_to_pydantic_field(opt: click.Option) -> tuple:
    """
    Map a click.Option to a Pydantic field tuple: (type, default)
    The machine dissects Click options and rebuilds them as Pydantic fields.
    """
    field_name = opt.opts[-1].lstrip("-").replace("-", "_")

    # Handle flags - the simplest case
    if opt.is_flag:
        return (bool, opt.default if opt.default is not None else False)

    # Handle multiple values - arrays in JSON
    if opt.multiple:
        base_type = _get_base_type(opt.type)
        return (List[base_type], opt.default or [])

    # Handle choices - constrained strings
    if hasattr(opt.type, 'choices') and opt.type.choices:
        return (str, opt.default)

    # Handle file types - special case for uploads
    if isinstance(opt.type, click.File):
        return (Optional[str], None)  # File path as string

    # Handle path types
    if isinstance(opt.type, click.Path):
        return (str, opt.default)

    # Handle tuple types
    if isinstance(opt.type, click.Tuple):
        return (List[str], opt.default or [])

    # Map basic types
    type_map = {
        'IntParamType': int,
        'FloatParamType': float,
        'BoolParamType': bool,
        'StringParamType': str,
    }

    type_name = type(opt.type).__name__
    python_type = type_map.get(type_name, str)

    return (python_type, opt.default)

def _get_base_type(click_type) -> Type:
    """Extract the base type from a Click type for list elements."""
    if hasattr(click_type, 'param_type'):
        return _get_base_type(click_type.param_type)

    type_map = {
        'IntParamType': int,
        'FloatParamType': float,
        'BoolParamType': bool,
        'StringParamType': str,
    }

    return type_map.get(type(click_type).__name__, str)

def _build_model_from_click_params(name: str, params: List[click.Parameter]) -> Type[BaseModel]:
    """
    Build a Pydantic model from Click parameters.
    The machine reconstructs CLI interfaces as HTTP request models.
    """
    fields = {}

    for param in params:
        if isinstance(param, click.Option):
            py_type, default = _click_option_to_pydantic_field(param)
            field_name = param.opts[-1].lstrip("-").replace("-", "_")

            # Handle required fields
            if param.required and default is None:
                fields[field_name] = (py_type, ...)
            else:
                fields[field_name] = (py_type, default)

        elif isinstance(param, click.Argument):
            # Arguments become required fields
            field_name = param.name
            py_type = _get_base_type(param.type)
            fields[field_name] = (py_type, ...)

    # Create the model dynamically
    Model = create_model(f"{name}_model", **fields)
    return Model

def _determine_http_method(cmd: click.Command) -> str:
    """
    Determine HTTP method based on command semantics.
    The machine reads between the lines of command names.
    """
    cmd_name = cmd.name.lower()

    # Destructive operations
    if any(verb in cmd_name for verb in ['delete', 'remove', 'destroy', 'kill']):
        return 'DELETE'

    # Update operations
    if any(verb in cmd_name for verb in ['update', 'modify', 'edit', 'patch']):
        return 'PATCH'

    # Create operations
    if any(verb in cmd_name for verb in ['create', 'add', 'new', 'build', 'deploy']):
        return 'POST'

    # Read operations
    if any(verb in cmd_name for verb in ['list', 'show', 'get', 'status', 'info']):
        return 'GET'

    # Default to POST for commands with side effects
    return 'POST'

def _build_path_from_command(cmd: click.Command, parent_path: str = "") -> str:
    """
    Build HTTP path from command hierarchy.
    The machine maps CLI structure to URL structure.
    """
    cmd_name = cmd.name.replace("_", "-")

    if parent_path:
        return f"{parent_path}/{cmd_name}"
    else:
        return f"/{cmd_name}"

def _handle_file_uploads(params: Dict[str, Any], files: Dict[str, UploadFile]) -> Dict[str, Any]:
    """
    Handle file uploads by writing to temporary files.
    The machine bridges the gap between HTTP uploads and CLI file handling.
    """
    processed_params = params.copy()

    for field_name, upload_file in files.items():
        if upload_file:
            # Create temporary file
            with tempfile.NamedTemporaryFile(delete=False, suffix=Path(upload_file.filename).suffix) as tmp:
                content = upload_file.file.read()
                tmp.write(content)
                tmp.flush()
                processed_params[field_name] = tmp.name

    return processed_params

def _run_click_command(cmd: click.Command, params: Dict[str, Any]) -> Any:
    """
    Execute Click command with proper context and error handling.
    The machine preserves Click's execution semantics.
    """
    try:
        # Create Click context
        ctx = click.Context(cmd)

        # Populate context parameters
        ctx.params.update(params)

        # Execute command using Click's invoke mechanism
        result = cmd.invoke(ctx)
        return result

    except click.ClickException as e:
        # Convert Click exceptions to HTTP exceptions
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        # Handle unexpected errors
        raise HTTPException(status_code=500, detail="Internal server error")

def click_to_router(
    click_root: click.BaseCommand, 
    prefix: str = "",
    method_map: Optional[Dict[str, str]] = None,
    auth_map: Optional[Dict[str, str]] = None,
    background_commands: Optional[List[str]] = None,
    streaming_commands: Optional[List[str]] = None
) -> APIRouter:
    """
    Convert Click command hierarchy to FastAPI router.
    The machine's primary function - where CLI meets HTTP.
    """
    router = APIRouter(prefix=prefix)
    method_map = method_map or {}
    auth_map = auth_map or {}
    background_commands = background_commands or []
    streaming_commands = streaming_commands or []

    if isinstance(click_root, click.Group):
        # Handle command groups - recursive descent
        for name, cmd in click_root.commands.items():
            subpath = _build_path_from_command(cmd, prefix)

            if isinstance(cmd, click.Group):
                # Recursively process sub-groups
                subrouter = click_to_router(cmd, prefix=subpath, method_map=method_map)
                router.include_router(subrouter)
            else:
                # Process individual commands
                _add_command_endpoint(router, cmd, subpath, method_map, auth_map, 
                                    background_commands, streaming_commands)
    else:
        # Handle single command at root
        _add_command_endpoint(router, click_root, "/", method_map, auth_map, 
                            background_commands, streaming_commands)

    return router

def _add_command_endpoint(
    router: APIRouter,
    cmd: click.Command,
    path: str,
    method_map: Dict[str, str],
    auth_map: Dict[str, str],
    background_commands: List[str],
    streaming_commands: List[str]
):
    """
    Add a single command as an HTTP endpoint.
    The machine's surgical precision in endpoint creation.
    """
    # Determine HTTP method
    method = method_map.get(cmd.name, _determine_http_method(cmd))

    # Build Pydantic model for request validation
    Model = _build_model_from_click_params(cmd.name, cmd.params)

    # Check if command should run in background
    is_background = cmd.name in background_commands

    # Check if command should stream output
    is_streaming = cmd.name in streaming_commands

    # Build endpoint handler
    async def endpoint_handler(
        payload: Model = Depends(Model),
        background_tasks: BackgroundTasks = Depends(BackgroundTasks),
        files: Dict[str, UploadFile] = File(None)
    ):
        # Convert Pydantic model to dict
        params = payload.dict()

        # Handle file uploads
        if files:
            params = _handle_file_uploads(params, files)

        # Handle authentication if mapped
        if cmd.name in auth_map:
            # This would be implemented with proper auth dependency
            pass

        # Execute command
        if is_background:
            # Run in background task
            background_tasks.add_task(_run_click_command, cmd, params)
            return {"status": "accepted", "message": "Command running in background"}

        elif is_streaming:
            # Stream command output
            return StreamingResponse(
                _stream_command_output(cmd, params),
                media_type="text/plain"
            )

        else:
            # Run synchronously in threadpool
            result = await run_in_threadpool(_run_click_command, cmd, params)
            return {"result": result}

    # Register endpoint
    router.add_api_route(
        path,
        endpoint_handler,
        methods=[method],
        name=cmd.name,
        summary=cmd.help or f"Execute {cmd.name} command"
    )

def _stream_command_output(cmd: click.Command, params: Dict[str, Any]):
    """
    Stream command output for long-running commands.
    The machine's solution to the streaming problem.
    """
    def generate():
        # This is a simplified version - in practice, you'd need to
        # capture stdout/stderr from the command execution
        try:
            result = _run_click_command(cmd, params)
            if result:
                yield str(result).encode()
        except Exception as e:
            yield f"Error: {str(e)}".encode()

    return generate()

# app.py - The machine's host
from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import click
from factory import click_to_router

# Import your Click CLI
from cli import cli  # Your Click CLI module

app = FastAPI(
    title="Click CLI to FastAPI Converter",
    description="Automatically generated API from Click CLI",
    version="1.0.0"
)

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Convert Click CLI to FastAPI router
api_router = click_to_router(
    cli,
    prefix="/api",
    method_map={
        "build": "POST",
        "deploy": "POST",
        "status": "GET",
        "logs": "GET"
    },
    background_commands=["build", "deploy"],
    streaming_commands=["logs", "tail"]
)

# Include the generated router
app.include_router(api_router)

# Add health check
@app.get("/health")
async def health_check():
    return {"status": "healthy", "message": "Click CLI converter is running"}

# Add OpenAPI documentation
@app.get("/")
async def root():
    return {
        "message": "Click CLI to FastAPI Converter",
        "docs": "/docs",
        "openapi": "/openapi.json"
    }

# cli.py - The machine's input
import click
import json
import time
from pathlib import Path

@click.group()
def cli():
    """Top-level CLI group"""
    pass

@cli.command()
@click.argument("project")
@click.option("--verbose", "-v", is_flag=True, help="Verbose output")
@click.option("--config", "-c", help="Configuration file")
@click.option("--tags", multiple=True, help="Build tags")
def build(project, verbose, config, tags):
    """Build a project"""
    if verbose:
        click.echo(f"Building {project} with verbose output")

    if config:
        click.echo(f"Using config: {config}")

    if tags:
        click.echo(f"Tags: {', '.join(tags)}")

    # Simulate build process
    time.sleep(1)

    result = {
        "project": project,
        "status": "success",
        "tags": list(tags) if tags else [],
        "config": config
    }

    click.echo(json.dumps(result, indent=2))
    return result

@cli.command()
@click.argument("service")
@click.option("--port", type=int, default=8000, help="Port number")
@click.option("--workers", type=int, default=1, help="Number of workers")
@click.option("--reload", is_flag=True, help="Enable auto-reload")
def deploy(service, port, workers, reload):
    """Deploy a service"""
    click.echo(f"Deploying {service} on port {port}")
    click.echo(f"Workers: {workers}")
    click.echo(f"Reload: {reload}")

    # Simulate deployment
    time.sleep(2)

    return {
        "service": service,
        "port": port,
        "workers": workers,
        "reload": reload,
        "status": "deployed"
    }

@cli.command()
@click.option("--service", help="Service name")
def status(service):
    """Get service status"""
    if service:
        click.echo(f"Status of {service}: running")
        return {"service": service, "status": "running"}
    else:
        click.echo("All services: running")
        return {"services": ["web", "api", "db"], "status": "running"}

@cli.command()
@click.argument("file", type=click.Path(exists=True))
@click.option("--lines", "-n", type=int, default=10, help="Number of lines")
def tail(file, lines):
    """Tail a file"""
    path = Path(file)
    click.echo(f"Tailing {path} (last {lines} lines)")

    # Simulate tailing
    for i in range(lines):
        click.echo(f"Line {i+1}: Sample log entry")
        time.sleep(0.1)

    return {"file": str(path), "lines": lines}

@cli.group()
def user():
    """User management commands"""
    pass

@user.command()
@click.argument("username")
@click.option("--email", help="User email")
@click.option("--admin", is_flag=True, help="Make user admin")
def add(username, email, admin):
    """Add a new user"""
    click.echo(f"Adding user: {username}")
    if email:
        click.echo(f"Email: {email}")
    if admin:
        click.echo("Admin privileges: yes")

    return {
        "username": username,
        "email": email,
        "admin": admin,
        "status": "created"
    }

@user.command()
@click.argument("username")
def remove(username):
    """Remove a user"""
    click.echo(f"Removing user: {username}")
    return {"username": username, "status": "removed"}

if __name__ == "__main__":
    cli()

# tests/test_factory.py - The machine's self-diagnosis
import pytest
from fastapi.testclient import TestClient
from factory import _click_option_to_pydantic_field, _build_model_from_click_params
import click

def test_click_option_to_pydantic_field():
    """Test Click option to Pydantic field conversion"""
    # Test flag option
    flag_opt = click.Option(['--verbose'], is_flag=True, default=False)
    py_type, default = _click_option_to_pydantic_field(flag_opt)
    assert py_type == bool
    assert default == False

    # Test string option
    str_opt = click.Option(['--name'], type=str, default='default')
    py_type, default = _click_option_to_pydantic_field(str_opt)
    assert py_type == str
    assert default == 'default'

    # Test multiple option
    multi_opt = click.Option(['--tags'], multiple=True, type=str)
    py_type, default = _click_option_to_pydantic_field(multi_opt)
    assert py_type == list
    assert default == []

def test_build_model_from_click_params():
    """Test Pydantic model generation from Click parameters"""
    # Create test command
    @click.command()
    @click.option('--name', type=str, required=True)
    @click.option('--age', type=int, default=25)
    @click.option('--verbose', is_flag=True)
    def test_cmd(name, age, verbose):
        pass

    # Generate model
    Model = _build_model_from_click_params('test', test_cmd.params)

    # Test model creation
    instance = Model(name='Alice', age=30, verbose=True)
    assert instance.name == 'Alice'
    assert instance.age == 30
    assert instance.verbose == True

    # Test required field validation
    with pytest.raises(ValueError):
        Model(age=30)  # Missing required 'name' field

def test_http_method_determination():
    """Test HTTP method determination"""
    from factory import _determine_http_method

    # Test delete command
    @click.command(name='delete')
    def delete_cmd():
        pass

    assert _determine_http_method(delete_cmd) == 'DELETE'

    # Test build command
    @click.command(name='build')
    def build_cmd():
        pass

    assert _determine_http_method(build_cmd) == 'POST'

    # Test status command
    @click.command(name='status')
    def status_cmd():
        pass

    assert _determine_http_method(status_cmd) == 'GET'

# tests/test_integration.py - The machine's end-to-end validation
import pytest
from fastapi.testclient import TestClient
from app import app

client = TestClient(app)

def test_build_endpoint():
    """Test build command endpoint"""
    response = client.post(
        "/api/build",
        json={
            "project": "test-project",
            "verbose": True,
            "tags": ["v1.0", "production"]
        }
    )

    assert response.status_code == 200
    data = response.json()
    assert data["result"]["project"] == "test-project"
    assert data["result"]["status"] == "success"

def test_deploy_endpoint():
    """Test deploy command endpoint"""
    response = client.post(
        "/api/deploy",
        json={
            "service": "web",
            "port": 8080,
            "workers": 4,
            "reload": True
        }
    )

    assert response.status_code == 200
    data = response.json()
    assert data["result"]["service"] == "web"
    assert data["result"]["port"] == 8080

def test_status_endpoint():
    """Test status command endpoint"""
    response = client.get("/api/status")

    assert response.status_code == 200
    data = response.json()
    assert "services" in data["result"]

def test_user_add_endpoint():
    """Test user add command endpoint"""
    response = client.post(
        "/api/user/add",
        json={
            "username": "alice",
            "email": "alice@example.com",
            "admin": True
        }
    )

    assert response.status_code == 200
    data = response.json()
    assert data["result"]["username"] == "alice"
    assert data["result"]["admin"] == True

def test_user_remove_endpoint():
    """Test user remove command endpoint"""
    response = client.delete(
        "/api/user/remove",
        json={"username": "alice"}
    )

    assert response.status_code == 200
    data = response.json()
    assert data["result"]["username"] == "alice"

def test_validation_error():
    """Test input validation"""
    response = client.post(
        "/api/build",
        json={"project": "test"}  # Missing required fields
    )

    # Should still work if only required fields are provided
    assert response.status_code == 200

def test_health_check():
    """Test health check endpoint"""
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

# dependencies.py - The machine's support infrastructure
from fastapi import Depends, HTTPException, Header
from typing import Optional
import os

def get_current_user(authorization: Optional[str] = Header(None)):
    """
    Extract user from Authorization header.
    The machine's authentication mechanism.
    """
    if not authorization:
        raise HTTPException(status_code=401, detail="Authorization header required")

    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid authorization format")

    token = authorization[7:]  # Remove "Bearer " prefix

    # Simple token validation (replace with proper JWT validation)
    if token != "valid-token":
        raise HTTPException(status_code=401, detail="Invalid token")

    return {"user_id": "user123", "username": "admin"}

def get_config():
    """
    Load configuration for CLI context.
    The machine's configuration management.
    """
    return {
        "database_url": os.getenv("DATABASE_URL", "sqlite:///app.db"),
        "debug": os.getenv("DEBUG", "false").lower() == "true",
        "log_level": os.getenv("LOG_LEVEL", "INFO")
    }

def get_cli_context():
    """
    Recreate Click context for command execution.
    The machine's context preservation.
    """
    return {
        "config": get_config(),
        "user": get_current_user()
    }

# Enhanced file handling in factory.py
def _handle_file_uploads_advanced(params: Dict[str, Any], files: Dict[str, UploadFile]) -> Dict[str, Any]:
    """
    Advanced file upload handling.
    The machine's solution to the file upload problem.
    """
    processed_params = params.copy()

    for field_name, upload_file in files.items():
        if upload_file:
            # Determine if file should be saved or passed as path
            if field_name.endswith('_path'):
                # Save to temporary file and pass path
                with tempfile.NamedTemporaryFile(delete=False, suffix=Path(upload_file.filename).suffix) as tmp:
                    content = upload_file.file.read()
                    tmp.write(content)
                    tmp.flush()
                    processed_params[field_name] = tmp.name
            else:
                # Pass file object directly
                processed_params[field_name] = upload_file.file

    return processed_params

# Enhanced streaming in factory.py
def _stream_command_output_advanced(cmd: click.Command, params: Dict[str, Any]):
    """
    Advanced streaming for command output.
    The machine's solution to the streaming problem.
    """
    import subprocess
    import sys

    def generate():
        try:
            # Convert command to subprocess call
            cmd_args = [sys.executable, "-m", "cli", cmd.name]

            # Add parameters
            for key, value in params.items():
                if isinstance(value, bool) and value:
                    cmd_args.append(f"--{key}")
                elif not isinstance(value, bool):
                    cmd_args.extend([f"--{key}", str(value)])

            # Run subprocess and stream output
            process = subprocess.Popen(
                cmd_args,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                bufsize=1,
                universal_newlines=True
            )

            for line in iter(process.stdout.readline, ''):
                yield line.encode()

            process.wait()

        except Exception as e:
            yield f"Error: {str(e)}".encode()

    return generate()