Add codegen tool and exception hierarchy

- Added tools/codegen.py to generate safe WIT bindings.
- Added fastly_compute/exceptions.py to define the exception hierarchy.
- Updated Makefile to include codegen target.

Author: posborne

Makefile

@@ -75,7 +75,11 @@ build-all: all
 
 # Clean build artifacts
 clean:
-	rm -rf $(BUILD_DIR) $(STUBS_DIR)
+	rm -rf $(BUILD_DIR) $(STUBS_DIR) fastly_compute/wit
+
+# Code Generation
+codegen: $(STUBS_DIR)
+	uv run python3 tools/codegen.py
 
 # Development tools
 lint: | $(STUBS_DIR)

fastly_compute/exceptions.py

@@ -0,0 +1,73 @@
+from typing import Any
+
+
+class FastlyError(Exception):
+    """Base class for all Fastly Compute exceptions."""
+
+    def __init__(self, message: str, wit_error: Any | None = None):
+        super().__init__(message)
+        self.wit_error = wit_error
+
+
+class ResourceError(FastlyError):
+    """Resource open/access errors."""
+
+    pass
+
+
+class ResourceOpenError(ResourceError):
+    """Error opening a resource."""
+
+    pass
+
+
+class ResourceNotFound(ResourceError):
+    """Resource not found."""
+
+    pass
+
+
+class ResourceLimitExceeded(ResourceError):
+    """Quotas or limits exceeded."""
+
+    pass
+
+
+class BackendError(FastlyError):
+    """Backend communication errors."""
+
+    pass
+
+
+class BadRequestError(FastlyError):
+    """Bad Request."""
+
+    pass
+
+
+class RateLimitExceeded(FastlyError):
+    """Rate limit exceeded."""
+
+    pass
+
+
+# KV Store Specific
+class KVStoreError(FastlyError):
+    pass
+
+
+class KVKeyFound(KVStoreError):
+    pass
+
+
+class KVPreconditionFailed(KVStoreError):
+    pass
+
+
+class KVPayloadTooLarge(KVStoreError):
+    pass
+
+
+# ACL Specific
+class ACLError(FastlyError):
+    pass

tools/codegen.py

@@ -0,0 +1,280 @@
+#!/usr/bin/env python3
+import json
+import os
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+# Configuration
+WIT_DIR = Path("wit")
+OUTPUT_DIR = Path("fastly_compute/wit")
+RAW_BINDINGS_PACKAGE = "wit_world"  # For now, use existing location
+
+# Mapping from WIT Error Enum Names to Base Exception Classes
+# and specific variants to specific exceptions.
+# Format: "wit-enum-name": ("BaseException", {"variant": "SpecificException"})
+ERROR_MAPPING = {
+    "kv-error": (
+        "KVStoreError",
+        {
+            "bad-request": "BadRequestError",
+            "precondition-failed": "KVPreconditionFailed",
+            "payload-too-large": "KVPayloadTooLarge",
+            "too-many-requests": "RateLimitExceeded",
+            "internal-error": "BackendError",
+        },
+    ),
+    "acl-error": (
+        "ACLError",
+        {
+            "too-many-requests": "RateLimitExceeded",
+        },
+    ),
+    "open-error": (
+        "ResourceOpenError",
+        {
+            "name-empty": "ValueError",
+            "name-too-long": "ValueError",
+            "name-contains-invalid-char": "ValueError",
+            "file-not-found": "ResourceNotFound",  # Assuming implicit
+        },
+    ),
+    "error": ("FastlyError", {}),
+    "error-with-detail": ("FastlyError", {}),  # Generic error with string
+}
+
+
+def to_snake_case(name: str) -> str:
+    return name.replace("-", "_")
+
+
+def to_camel_case(name: str) -> str:
+    parts = name.split("-")
+    return "".join(p.capitalize() for p in parts)
+
+
+def run_wasm_tools(wit_dir: Path) -> dict[str, Any]:
+    """Run wasm-tools to get JSON representation of WIT."""
+    try:
+        result = subprocess.run(
+            ["wasm-tools", "component", "wit", str(wit_dir), "--json"],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        return json.loads(result.stdout)
+    except subprocess.CalledProcessError as e:
+        print(f"Error running wasm-tools: {e.stderr}", file=sys.stderr)
+        sys.exit(1)
+    except FileNotFoundError:
+        print("wasm-tools not found in PATH", file=sys.stderr)
+        sys.exit(1)
+
+
+def resolve_type(types: list[dict], type_id: int) -> dict:
+    if type_id < len(types):
+        return types[type_id]
+    return {}
+
+
+def find_error_type(types: list[dict], type_id: Any) -> str | None:
+    """Trace a type ID to find if it's a Result with an Error Enum."""
+    if isinstance(type_id, int):
+        t = resolve_type(types, type_id)
+        kind = t.get("kind", {})
+        if "result" in kind:
+            result = kind["result"]
+            err_id = result.get("err")
+            if err_id is not None:
+                err_type = resolve_type(types, err_id)
+                # Check if it has a name (Enum or Resource)
+                return err_type.get("name")
+        elif "type" in kind:
+            # Alias
+            return find_error_type(types, kind["type"])
+    return None
+
+
+def generate_utils_module(output_dir: Path) -> None:
+    content = """
+from functools import wraps
+from typing import Any, Callable, Dict, Type, Optional
+from fastly_compute.witraw.imports.types import Err
+from fastly_compute.exceptions import FastlyError
+
+def map_wit_error(mapping: Dict[str, Type[FastlyError]], default: Type[FastlyError] = FastlyError):
+    def decorator(func: Callable) -> Callable:
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            try:
+                return func(*args, **kwargs)
+            except Err as e:
+                err_val = e.value
+                variant = str(err_val)
+                exc_cls = mapping.get(variant)
+                if exc_cls:
+                    raise exc_cls(variant) from e
+                raise default(f"Generic error: {err_val}") from e
+        return wrapper
+    return decorator
+"""
+    with open(output_dir / "utils.py", "w") as f:
+        f.write(content.strip())
+
+
+def generate_module(interface: dict, types: list[dict], module_name: str) -> str:
+    functions = interface.get("functions", {})
+    if not functions:
+        return ""
+
+    lines = []
+    lines.append("# Generated by tools/codegen.py. DO NOT EDIT.")
+    lines.append(f"from fastly_compute.witraw.imports import {module_name} as _raw")
+    lines.append("from fastly_compute.wit.utils import map_wit_error")
+    lines.append("from fastly_compute.exceptions import *")
+    lines.append("")
+
+    # Determine error mapping for this module
+    # We scan all functions to find the common error enum(s)
+    # Ideally, an interface uses one main error type (e.g. kv-error)
+    # If multiple, we might need multiple maps or a merged one.
+
+    # Collect all error types used in this module
+    error_types = set()
+    for func_def in functions.values():
+        res = func_def.get("result")
+        err_name = find_error_type(types, res)
+        if err_name:
+            error_types.add(err_name)
+
+    # Generate Mappings
+    for err_name in error_types:
+        if err_name in ERROR_MAPPING:
+            base_exc, variants = ERROR_MAPPING[err_name]
+            # Generate dict definition
+            lines.append(f"_{to_snake_case(err_name).upper()}_MAP = {{")
+            for var, exc in variants.items():
+                lines.append(f"    '{var}': {exc},")
+            lines.append("}")
+            lines.append("")
+
+    # Group functions by resource (same as before)
+    resources = {}
+    freestanding = []
+
+    for func_key, func_def in functions.items():
+        name = func_def["name"]
+        if "." in name and not name.startswith(
+            "["
+        ):  # Handle [method] prefix removal logic
+            pass  # Logic handles below
+
+        # Robust parsing of "[kind]resource.method"
+        clean_name = re.sub(r"\[.*?\]", "", name)
+        if "." in clean_name:
+            res_name, method_name = clean_name.split(".", 1)
+            if res_name not in resources:
+                resources[res_name] = []
+            resources[res_name].append((method_name, func_def))
+        else:
+            freestanding.append((name, func_def))
+
+    # Helper to get decorator string
+    def get_decorator(func_def):
+        res = func_def.get("result")
+        err_name = find_error_type(types, res)
+        if err_name and err_name in ERROR_MAPPING:
+            map_name = f"_{to_snake_case(err_name).upper()}_MAP"
+            base_exc = ERROR_MAPPING[err_name][0]
+            return f"@map_wit_error({map_name}, default={base_exc})"
+        return None
+
+    # Generate Freestanding
+    for name, func_def in freestanding:
+        py_name = to_snake_case(name)
+        deco = get_decorator(func_def)
+        if deco:
+            lines.append(deco)
+        lines.append(f"def {py_name}(*args, **kwargs):")
+        lines.append(f"    return _raw.{to_snake_case(name)}(*args, **kwargs)")
+        lines.append("")
+
+    # Generate Classes
+    for res_name, methods in resources.items():
+        class_name = to_camel_case(res_name)
+        lines.append(f"class {class_name}:")
+        lines.append("    def __init__(self, handle):")
+        lines.append("        self._handle = handle")
+        lines.append("")
+
+        for name, func_def in methods:
+            py_name = to_snake_case(name)
+            deco = get_decorator(func_def)
+            if deco:
+                lines.append(f"    {deco}")
+
+            # Check static vs method
+            is_static = "[static]" in func_def["name"]
+
+            if is_static:
+                # Static method
+                lines.append("    @classmethod")
+                lines.append(f"    def {py_name}(cls, *args, **kwargs):")
+                lines.append(
+                    f"        return _raw.{class_name}.{to_snake_case(name)}(*args, **kwargs)"
+                )
+            else:
+                # Instance method
+                lines.append(f"    def {py_name}(self, *args, **kwargs):")
+                # Pass self._handle as first arg if raw expects it?
+                # Raw bindings for methods usually take 'self' handle as first arg.
+                lines.append(
+                    f"        return _raw.{class_name}.{to_snake_case(name)}(self._handle, *args, **kwargs)"
+                )
+            lines.append("")
+
+    return "\n".join(lines)
+
+
+def main():
+    # Ensure output dir
+    if not OUTPUT_DIR.exists():
+        os.makedirs(OUTPUT_DIR)
+
+    imports_dir = OUTPUT_DIR / "imports"
+    if not imports_dir.exists():
+        os.makedirs(imports_dir)
+
+    (OUTPUT_DIR / "__init__.py").touch()
+    (imports_dir / "__init__.py").touch()
+
+    # Generate utils
+    generate_utils_module(OUTPUT_DIR)
+
+    print(f"Parsing WIT from {WIT_DIR}...")
+    data = run_wasm_tools(WIT_DIR)
+
+    types = data.get("types", [])
+    interfaces = data.get("interfaces", [])
+
+    for interface in interfaces:
+        name = interface.get("name")
+        if not name:
+            continue
+
+        print(f"Generating wrapper for {name}...")
+        py_mod_name = to_snake_case(name)
+
+        content = generate_module(interface, types, py_mod_name)
+
+        if content:
+            with open(imports_dir / f"{py_mod_name}.py", "w") as f:
+                f.write(content)
+
+    print("Done.")
+
+
+if __name__ == "__main__":
+    main()