Add upload-effect and effects commands

- artdag upload-effect: upload effect files with metadata parsing
- artdag effects: list uploaded effects

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gilesb
2026-01-12 06:37:24 +00:00
parent e642ffe301
commit fa540f4441

127
artdag.py
View File

@@ -944,27 +944,52 @@ def collection_delete(name):
# ============ Recipe Commands ============
def _is_sexp_file(filepath: str, content: str) -> bool:
"""Detect if file is S-expression format."""
# Check extension first
if filepath.endswith('.sexp'):
return True
# Check content - skip comments and whitespace
for line in content.split('\n'):
stripped = line.strip()
if not stripped or stripped.startswith(';'):
continue
return stripped.startswith('(')
return False
@cli.command("upload-recipe")
@click.argument("filepath", type=click.Path(exists=True))
def upload_recipe(filepath):
"""Upload a recipe YAML file. Requires login."""
"""Upload a recipe file (YAML or S-expression). Requires login."""
token_data = load_token()
if not token_data.get("access_token"):
click.echo("Not logged in. Please run: artdag login <username>", err=True)
sys.exit(1)
# Validate YAML locally first
# Read content
with open(filepath) as f:
content = f.read()
# Detect format and validate
is_sexp = _is_sexp_file(filepath, content)
if is_sexp:
# S-expression - basic syntax check (starts with paren after comments)
# Full validation happens on server
click.echo("Detected S-expression format")
else:
# Validate YAML locally
try:
recipe = yaml.safe_load(f)
recipe = yaml.safe_load(content)
except yaml.YAMLError as e:
click.echo(f"Invalid YAML: {e}", err=True)
sys.exit(1)
# Check required fields
if not recipe.get("name"):
click.echo("Recipe must have a 'name' field", err=True)
sys.exit(1)
# Check required fields for YAML
if not recipe.get("name"):
click.echo("Recipe must have a 'name' field", err=True)
sys.exit(1)
# Upload
try:
@@ -975,6 +1000,8 @@ def upload_recipe(filepath):
if resp.status_code == 401:
click.echo("Authentication failed. Please login again.", err=True)
sys.exit(1)
if resp.status_code >= 400:
click.echo(f"Error response: {resp.text}", err=True)
resp.raise_for_status()
result = resp.json()
@@ -987,6 +1014,87 @@ def upload_recipe(filepath):
sys.exit(1)
@cli.command("upload-effect")
@click.argument("filepath", type=click.Path(exists=True))
def upload_effect(filepath):
"""Upload an effect file. Requires login.
Effects are Python files with PEP 723 dependencies and @-tag metadata.
Returns the content hash for use in recipes.
"""
token_data = load_token()
if not token_data.get("access_token"):
click.echo("Not logged in. Please run: artdag login <username>", err=True)
sys.exit(1)
# Check it's a Python file
if not filepath.endswith(".py"):
click.echo("Effect must be a Python file (.py)", err=True)
sys.exit(1)
# Upload
try:
with open(filepath, "rb") as f:
files = {"file": (Path(filepath).name, f)}
headers = {"Authorization": f"Bearer {token_data['access_token']}"}
resp = requests.post(f"{get_server()}/effects/upload", files=files, headers=headers)
if resp.status_code == 401:
click.echo("Authentication failed. Please login again.", err=True)
sys.exit(1)
if resp.status_code >= 400:
click.echo(f"Error response: {resp.text}", err=True)
sys.exit(1)
resp.raise_for_status()
result = resp.json()
click.echo(f"Uploaded effect: {result['name']} v{result.get('version', '1.0.0')}")
click.echo(f"Content hash: {result['content_hash']}")
click.echo(f"Temporal: {result.get('temporal', False)}")
if result.get('params'):
click.echo(f"Parameters: {', '.join(p['name'] for p in result['params'])}")
if result.get('dependencies'):
click.echo(f"Dependencies: {', '.join(result['dependencies'])}")
click.echo()
click.echo("Use in recipes:")
click.echo(f' (effect {result["name"]} :hash "{result["content_hash"]}")')
except requests.RequestException as e:
click.echo(f"Upload failed: {e}", err=True)
sys.exit(1)
@cli.command("effects")
@click.option("--limit", "-l", default=20, help="Max effects to show")
def list_effects(limit):
"""List uploaded effects."""
try:
headers = {}
token_data = load_token()
if token_data.get("access_token"):
headers["Authorization"] = f"Bearer {token_data['access_token']}"
resp = requests.get(f"{get_server()}/effects", headers=headers)
resp.raise_for_status()
result = resp.json()
effects = result.get("effects", [])[:limit]
if not effects:
click.echo("No effects found")
return
click.echo(f"Effects ({len(effects)}):\n")
for effect in effects:
meta = effect.get("meta", {})
click.echo(f" {meta.get('name', 'unknown')} v{meta.get('version', '?')}")
click.echo(f" Hash: {effect['content_hash'][:32]}...")
click.echo(f" Temporal: {meta.get('temporal', False)}")
if meta.get('params'):
click.echo(f" Params: {', '.join(p['name'] for p in meta['params'])}")
click.echo()
except requests.RequestException as e:
click.echo(f"Failed to list effects: {e}", err=True)
sys.exit(1)
@cli.command("recipes")
@click.option("--limit", "-l", default=10, help="Max recipes to show")
def list_recipes(limit):
@@ -1097,8 +1205,9 @@ def run_recipe(recipe_id, inputs, wait):
sys.exit(1)
click.echo(f"Run started: {result['run_id']}")
click.echo(f"Recipe: {result['recipe']}")
click.echo(f"Status: {result['status']}")
if result.get('recipe'):
click.echo(f"Recipe: {result['recipe']}")
click.echo(f"Status: {result.get('status', 'pending')}")
if wait:
click.echo("Waiting for completion...")