56009c391d
Testing setup: - Add pyproject.toml with mypy and pytest configuration - Add requirements-dev.txt for development dependencies - Create tests/ directory with test fixtures - Add 17 unit tests for DAG transformation pipeline Type annotations: - Add app/types.py with TypedDict definitions for node configs - Add typed helper functions: transform_node, build_input_name_mapping, bind_inputs, prepare_dag_for_execution - Refactor run_recipe to use the new typed helpers Regression tests for today's bugs: - test_effect_cid_key_not_effect_hash: Verifies CID uses 'cid' key - test_source_cid_binding_persists: Verifies bound CIDs in final DAG Run tests with: pytest tests/ -v Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
161 lines
3.9 KiB
Python
161 lines
3.9 KiB
Python
"""
|
|
Type definitions for Art DAG L1 server.
|
|
|
|
Uses TypedDict for configuration structures to enable mypy checking.
|
|
"""
|
|
|
|
from typing import Any, Dict, List, Optional, TypedDict, Union
|
|
from typing_extensions import NotRequired
|
|
|
|
|
|
# === Node Config Types ===
|
|
|
|
class SourceConfig(TypedDict, total=False):
|
|
"""Config for SOURCE nodes."""
|
|
cid: str # Content ID (IPFS CID or SHA3-256 hash)
|
|
asset: str # Asset name from registry
|
|
input: bool # True if this is a variable input
|
|
name: str # Human-readable name for variable inputs
|
|
description: str # Description for variable inputs
|
|
|
|
|
|
class EffectConfig(TypedDict, total=False):
|
|
"""Config for EFFECT nodes."""
|
|
effect: str # Effect name
|
|
cid: str # Effect CID (for cached/IPFS effects)
|
|
# Effect parameters are additional keys
|
|
intensity: float
|
|
level: float
|
|
|
|
|
|
class SequenceConfig(TypedDict, total=False):
|
|
"""Config for SEQUENCE nodes."""
|
|
transition: Dict[str, Any] # Transition config
|
|
|
|
|
|
class SegmentConfig(TypedDict, total=False):
|
|
"""Config for SEGMENT nodes."""
|
|
start: float
|
|
end: float
|
|
duration: float
|
|
|
|
|
|
# Union of all config types
|
|
NodeConfig = Union[SourceConfig, EffectConfig, SequenceConfig, SegmentConfig, Dict[str, Any]]
|
|
|
|
|
|
# === Node Types ===
|
|
|
|
class CompiledNode(TypedDict):
|
|
"""Node as produced by the S-expression compiler."""
|
|
id: str
|
|
type: str # "SOURCE", "EFFECT", "SEQUENCE", etc.
|
|
config: Dict[str, Any]
|
|
inputs: List[str]
|
|
name: NotRequired[str]
|
|
|
|
|
|
class TransformedNode(TypedDict):
|
|
"""Node after transformation for artdag execution."""
|
|
node_id: str
|
|
node_type: str
|
|
config: Dict[str, Any]
|
|
inputs: List[str]
|
|
name: NotRequired[str]
|
|
|
|
|
|
# === DAG Types ===
|
|
|
|
class CompiledDAG(TypedDict):
|
|
"""DAG as produced by the S-expression compiler."""
|
|
nodes: List[CompiledNode]
|
|
output: str
|
|
|
|
|
|
class TransformedDAG(TypedDict):
|
|
"""DAG after transformation for artdag execution."""
|
|
nodes: Dict[str, TransformedNode]
|
|
output_id: str
|
|
metadata: NotRequired[Dict[str, Any]]
|
|
|
|
|
|
# === Registry Types ===
|
|
|
|
class AssetEntry(TypedDict, total=False):
|
|
"""Asset in the recipe registry."""
|
|
cid: str
|
|
url: str
|
|
|
|
|
|
class EffectEntry(TypedDict, total=False):
|
|
"""Effect in the recipe registry."""
|
|
cid: str
|
|
url: str
|
|
temporal: bool
|
|
|
|
|
|
class Registry(TypedDict):
|
|
"""Recipe registry containing assets and effects."""
|
|
assets: Dict[str, AssetEntry]
|
|
effects: Dict[str, EffectEntry]
|
|
|
|
|
|
# === Recipe Types ===
|
|
|
|
class Recipe(TypedDict, total=False):
|
|
"""Compiled recipe structure."""
|
|
name: str
|
|
version: str
|
|
description: str
|
|
owner: str
|
|
registry: Registry
|
|
dag: CompiledDAG
|
|
recipe_id: str
|
|
ipfs_cid: str
|
|
sexp: str
|
|
step_count: int
|
|
error: str
|
|
|
|
|
|
# === API Request/Response Types ===
|
|
|
|
class RecipeRunInputs(TypedDict):
|
|
"""Mapping of input names to CIDs for recipe execution."""
|
|
# Keys are input names, values are CIDs
|
|
pass # Actually just Dict[str, str]
|
|
|
|
|
|
class RunResult(TypedDict, total=False):
|
|
"""Result of a recipe run."""
|
|
run_id: str
|
|
status: str # "pending", "running", "completed", "failed"
|
|
recipe: str
|
|
inputs: List[str]
|
|
output_cid: str
|
|
ipfs_cid: str
|
|
error: str
|
|
created_at: str
|
|
completed_at: str
|
|
|
|
|
|
# === Helper functions for type narrowing ===
|
|
|
|
def is_source_node(node: TransformedNode) -> bool:
|
|
"""Check if node is a SOURCE node."""
|
|
return node.get("node_type") == "SOURCE"
|
|
|
|
|
|
def is_effect_node(node: TransformedNode) -> bool:
|
|
"""Check if node is an EFFECT node."""
|
|
return node.get("node_type") == "EFFECT"
|
|
|
|
|
|
def is_variable_input(config: Dict[str, Any]) -> bool:
|
|
"""Check if a SOURCE node config represents a variable input."""
|
|
return bool(config.get("input"))
|
|
|
|
|
|
def get_effect_cid(config: Dict[str, Any]) -> Optional[str]:
|
|
"""Get effect CID from config, checking both 'cid' and 'hash' keys."""
|
|
return config.get("cid") or config.get("hash")
|