Server API (BlackD)

def main(bind_host: str, bind_port: int) -> None:
    """
    Start blackd HTTP server.

    Parameters:
    - bind_host: Host address to bind to (e.g., "127.0.0.1", "0.0.0.0")
    - bind_port: Port number to bind to (e.g., 45484)

    Note:
    - Runs asyncio event loop
    - Handles SIGINT/SIGTERM for graceful shutdown
    - Uses aiohttp web framework
    """

def make_app() -> web.Application:
    """
    Create aiohttp application with BlackD routes.

    Returns:
    Configured aiohttp.web.Application instance

    Note:
    - Configures POST / route for formatting
    - Sets up request handling and error responses
    - Includes CORS headers and protocol versioning
    """

def handle(request: web.Request, executor: Executor) -> web.Response:
    """
    Main HTTP request handler for formatting requests.

    Parameters:
    - request: aiohttp web request with Python code and headers
    - executor: ThreadPoolExecutor for async processing

    Returns:
    aiohttp web response with formatted code or error

    Request Format:
    - Method: POST
    - Path: /
    - Body: Python source code to format
    - Headers: Configuration options (see Header Constants)

    Response Format:
    - 200: Successfully formatted (body contains formatted code)
    - 204: No changes needed (empty body)
    - 400: Invalid request (syntax error, invalid headers)
    - 500: Internal formatting error

    Headers:
    - X-Black-Version: Black version used for formatting
    - Content-Type: text/plain for code, application/json for errors
    """

def parse_mode(headers: MultiMapping[str]) -> black.Mode:
    """
    Parse Black Mode configuration from HTTP headers.

    Parameters:
    - headers: HTTP request headers containing configuration

    Returns:
    Mode object with configuration from headers

    Supported Headers:
    - X-Line-Length: Maximum line length (integer)
    - X-Python-Variant: Target Python versions (comma-separated)
    - X-Skip-String-Normalization: "true" to disable string normalization
    - X-Skip-Magic-Trailing-Comma: "true" to disable trailing comma
    - X-Skip-Source-First-Line: "true" to skip first line
    - X-Preview: "true" to enable preview features
    - X-Unstable: "true" to enable unstable features
    - X-Enable-Unstable-Feature: Specific unstable features (comma-separated)

    Raises:
    - InvalidVariantHeader: If Python variant header is invalid
    - HeaderError: If other headers contain invalid values
    """

# Protocol and versioning
PROTOCOL_VERSION_HEADER = "X-Protocol-Version"
BLACK_VERSION_HEADER = "X-Black-Version"

# Configuration headers
LINE_LENGTH_HEADER = "X-Line-Length"
PYTHON_VARIANT_HEADER = "X-Python-Variant" 
SKIP_SOURCE_FIRST_LINE = "X-Skip-Source-First-Line"
SKIP_STRING_NORMALIZATION_HEADER = "X-Skip-String-Normalization"
SKIP_MAGIC_TRAILING_COMMA = "X-Skip-Magic-Trailing-Comma"
PREVIEW = "X-Preview"
UNSTABLE = "X-Unstable"
ENABLE_UNSTABLE_FEATURE = "X-Enable-Unstable-Feature"

# Processing options
FAST_OR_SAFE_HEADER = "X-Fast-Or-Safe"
DIFF_HEADER = "X-Diff"

class HeaderError(Exception):
    """Raised when HTTP headers contain invalid values."""
    pass

class InvalidVariantHeader(Exception):
    """Raised when X-Python-Variant header is invalid."""
    pass

import blackd

# Start server on localhost:45484
blackd.main("127.0.0.1", 45484)

# Start server on all interfaces
blackd.main("0.0.0.0", 8080)

# Default configuration
blackd

# Custom host and port  
blackd --bind-host 0.0.0.0 --bind-port 8080

import requests

# Format code via HTTP API
code_to_format = '''
def hello(name):
    print(f"Hello {name}!")
'''

response = requests.post(
    "http://localhost:45484/",
    data=code_to_format,
    headers={
        "Content-Type": "text/plain",
        "X-Line-Length": "79",
        "X-Python-Variant": "py39,py310", 
        "X-Skip-String-Normalization": "false",
        "X-Fast-Or-Safe": "safe"
    }
)

if response.status_code == 200:
    formatted_code = response.text
    print("Formatted successfully:")
    print(formatted_code)
elif response.status_code == 204:
    print("No changes needed")
elif response.status_code == 400:
    print(f"Error: {response.text}")

# Format code with cURL
curl -X POST http://localhost:45484/ \
  -H "Content-Type: text/plain" \
  -H "X-Line-Length: 88" \
  -H "X-Python-Variant: py39" \
  -d "def hello(name):print(f'Hello {name}!')"

const axios = require('axios');

async function formatCode(code) {
    try {
        const response = await axios.post('http://localhost:45484/', code, {
            headers: {
                'Content-Type': 'text/plain',
                'X-Line-Length': '88',
                'X-Python-Variant': 'py39,py310',
                'X-Preview': 'true'
            }
        });
        
        if (response.status === 200) {
            return response.data; // Formatted code
        } else if (response.status === 204) {
            return code; // No changes needed
        }
    } catch (error) {
        if (error.response && error.response.status === 400) {
            throw new Error(`Formatting error: ${error.response.data}`);
        }
        throw error;
    }
}

// Usage
formatCode("def hello(name):print(f'Hello {name}!')")
    .then(formatted => console.log(formatted))
    .catch(error => console.error(error));

import requests
import json

class BlackDFormatter:
    def __init__(self, host="localhost", port=45484):
        self.base_url = f"http://{host}:{port}/"
        
    def format_code(self, code, **options):
        """Format code with optional configuration."""
        headers = {"Content-Type": "text/plain"}
        
        # Convert options to headers
        if "line_length" in options:
            headers["X-Line-Length"] = str(options["line_length"])
        if "target_versions" in options:
            headers["X-Python-Variant"] = ",".join(options["target_versions"])
        if "preview" in options and options["preview"]:
            headers["X-Preview"] = "true"
        if "skip_string_normalization" in options:
            headers["X-Skip-String-Normalization"] = str(options["skip_string_normalization"]).lower()
            
        try:
            response = requests.post(self.base_url, data=code, headers=headers, timeout=10)
            
            if response.status_code == 200:
                return response.text, True  # (formatted_code, changed)
            elif response.status_code == 204:
                return code, False  # (original_code, not_changed)  
            elif response.status_code == 400:
                raise ValueError(f"Syntax error: {response.text}")
            else:
                raise RuntimeError(f"Server error: {response.status_code}")
                
        except requests.RequestException as e:
            raise ConnectionError(f"Cannot connect to BlackD server: {e}")

# Usage in editor plugin
formatter = BlackDFormatter()

def format_selection(editor_code, line_length=88):
    try:
        formatted, changed = formatter.format_code(
            editor_code,
            line_length=line_length,
            target_versions=["py39", "py310"],
            preview=False
        )
        if changed:
            return formatted
        else:
            return None  # No changes needed
    except Exception as e:
        # Show error to user
        print(f"Formatting failed: {e}")
        return None

import aiohttp
import asyncio
import blackd

async def integrate_with_blackd():
    """Example of integrating BlackD into another aiohttp application."""
    
    # Create BlackD application
    blackd_app = blackd.make_app()
    
    # Create main application
    main_app = aiohttp.web.Application()
    
    # Add BlackD as a sub-application
    main_app.add_subapp('/format/', blackd_app)
    
    # Add other routes to main app
    async def health_check(request):
        return aiohttp.web.Response(text="OK")
    
    main_app.router.add_get('/health', health_check)
    
    return main_app

# Run integrated server
async def run_integrated_server():
    app = await integrate_with_blackd()
    runner = aiohttp.web.AppRunner(app)
    await runner.setup()
    
    site = aiohttp.web.TCPSite(runner, 'localhost', 8080)
    await site.start()
    
    print("Server running on http://localhost:8080")
    print("BlackD available at http://localhost:8080/format/")
    
    # Keep server running
    await asyncio.Future()  # Run forever

# Usage
# asyncio.run(run_integrated_server())

# Different formatting configurations via headers

# Strict Python 3.9 compatibility
headers = {
    "X-Python-Variant": "py39",
    "X-Line-Length": "79", 
    "X-Skip-String-Normalization": "false"
}

# Preview features enabled
headers = {
    "X-Preview": "true",
    "X-Line-Length": "100",
    "X-Python-Variant": "py310,py311"
}

# Conservative formatting
headers = {
    "X-Skip-String-Normalization": "true",
    "X-Skip-Magic-Trailing-Comma": "true",
    "X-Line-Length": "88"
}

# Unstable features
headers = {
    "X-Unstable": "true", 
    "X-Enable-Unstable-Feature": "string_processing,wrap_long_dict_values_in_parens"
}

import requests
import json

def format_with_error_handling(code):
    try:
        response = requests.post("http://localhost:45484/", data=code)
        
        if response.status_code == 200:
            return response.text
        elif response.status_code == 204:
            return code  # No changes
        elif response.status_code == 400:
            # Parse error details if JSON
            try:
                error_data = response.json()
                raise SyntaxError(error_data.get("message", response.text))
            except json.JSONDecodeError:
                raise SyntaxError(response.text)
        else:
            raise RuntimeError(f"Server error {response.status_code}: {response.text}")
            
    except requests.ConnectionError:
        raise ConnectionError("BlackD server not available")
    except requests.Timeout:
        raise TimeoutError("BlackD server timeout")

# aiohttp types
from aiohttp import web
from aiohttp.web import Request, Response, Application
from multidict import MultiMapping

# Async execution
from concurrent.futures import Executor, ThreadPoolExecutor

# Server configuration
ServerHost = str
ServerPort = int
HeaderDict = dict[str, str]