diff --git a/pyproject.toml b/pyproject.toml
index 384a760..5a44a48 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "py-aps"
-version = "0.0.4"
+version = "0.0.5"
description = "Autodesk Platform Service APIs Python SDK"
readme = "README.md"
requires-python = ">=3.9"
diff --git a/src/pyaps/automation/WORKFLOW.md b/src/pyaps/automation/WORKFLOW.md
new file mode 100644
index 0000000..6cffdd1
--- /dev/null
+++ b/src/pyaps/automation/WORKFLOW.md
@@ -0,0 +1,901 @@
+# AutomationWorkflow User Guide
+
+High-level workflow abstraction for executing Design Automation WorkItems.
+
+## Table of Contents
+
+- [Overview](#overview)
+- [Quick Start](#quick-start)
+- [Core Features](#core-features)
+ - [OSS Preparation](#1-oss-preparation)
+ - [WorkItem Execution](#2-workitem-execution)
+ - [Result Download](#3-result-download)
+ - [Unified Workflow](#4-unified-workflow)
+ - [Batch Processing](#5-batch-processing)
+- [Webhook Callbacks](#webhook-callbacks)
+- [Advanced Usage](#advanced-usage)
+- [Error Handling](#error-handling)
+- [Best Practices](#best-practices)
+
+---
+
+## Overview
+
+`AutomationWorkflow` provides a unified interface for Design Automation WorkItem execution:
+
+1. **OSS Bucket/Object Creation** - Prepare storage for input/output files
+2. **File Upload** - Upload local files to OSS and generate signed URLs
+3. **WorkItem Execution** - Execute Activity and monitor status
+4. **Result Download** - Download completed results to local filesystem
+
+---
+
+## Quick Start
+
+### 1. Initialize Clients
+
+```python
+from pyaps.auth import AuthClient, InMemoryTokenStore
+from pyaps.automation import AutomationClient, AutomationWorkflow, DEFAULT_AUTOMATION_SCOPES
+from pyaps.datamanagement import DataManagementClient
+
+# Authentication setup
+auth_client = AuthClient(
+ client_id="your_client_id",
+ client_secret="your_client_secret",
+ store=InMemoryTokenStore(),
+)
+
+def token_provider() -> str:
+ token = auth_client.two_legged.get_token(DEFAULT_AUTOMATION_SCOPES)
+ return token.access_token
+
+# Design Automation client
+auto_client = AutomationClient(
+ token_provider=token_provider,
+ region="us-east",
+)
+
+# Data Management client
+dm_client = DataManagementClient(
+ token_provider=token_provider,
+)
+
+# Initialize workflow
+workflow = AutomationWorkflow(
+ automation_client=auto_client,
+ data_client=dm_client,
+ default_bucket="my-design-automation-bucket",
+ poll_interval=10.0, # Check status every 10 seconds
+ timeout=3600.0, # Max wait time: 1 hour
+)
+```
+
+### 2. Simplest Usage
+
+```python
+# Execute entire workflow in one call
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={
+ "inputRvt": "path/to/input.rvt",
+ },
+ output_files={
+ "outputRvt": "output.rvt",
+ },
+ bucket_key="my-bucket",
+ download_outputs=True,
+ output_dir="./results",
+)
+
+print(f"Status: {result.status}")
+print(f"WorkItem ID: {result.workitem_id}")
+print(f"Report URL: {result.report_url}")
+```
+
+---
+
+## Core Features
+
+### 1. OSS Preparation
+
+#### 1.1 Create/Ensure Bucket
+
+```python
+bucket = workflow.ensure_bucket(
+ bucket_key="my-design-automation-bucket",
+ region="US", # US, EMEA, etc.
+ policy_key="transient", # transient (24h), temporary (30d), persistent (forever)
+)
+```
+
+#### 1.2 Upload Input File
+
+```python
+# Upload local file to OSS and get download URL
+input_url = workflow.upload_input_file(
+ local_path="path/to/input.rvt",
+ bucket_key="my-bucket",
+ object_key="inputs/input.rvt", # Optional, defaults to filename
+ timeout=300.0,
+)
+```
+
+#### 1.3 Prepare Output URL
+
+```python
+# Generate upload URL for output file
+output_url = workflow.prepare_output_url(
+ object_key="outputs/output.rvt",
+ bucket_key="my-bucket",
+ minutes_valid=60, # URL validity in minutes
+)
+```
+
+---
+
+### 2. WorkItem Execution
+
+#### 2.1 Start WorkItem
+
+```python
+workitem_id = workflow.start_workitem(
+ activity_id="myowner.RevitActivity+prod",
+ arguments={
+ "inputRvt": {"url": input_url, "verb": "get"},
+ "outputRvt": {"url": output_url, "verb": "put"},
+ },
+ nickname="MyWorkItem", # Optional
+)
+```
+
+#### 2.2 Wait for Completion
+
+```python
+def on_progress(status_data):
+ status = status_data.get("status")
+ progress = status_data.get("progress", "")
+ print(f"Status: {status} - {progress}")
+
+result = workflow.wait_for_completion(
+ workitem_id,
+ poll_interval=10.0,
+ timeout=3600.0,
+ on_progress=on_progress, # Optional callback
+)
+
+if result.status == "success":
+ print("WorkItem succeeded!")
+elif result.status == "failed":
+ print(f"WorkItem failed. Report: {result.report_url}")
+```
+
+#### 2.3 Cancel WorkItem
+
+```python
+workflow.cancel_workitem(workitem_id)
+```
+
+---
+
+### 3. Result Download
+
+```python
+workflow.download_output_file(
+ bucket_key="my-bucket",
+ object_key="outputs/output.rvt",
+ local_path="./results/output.rvt",
+)
+```
+
+---
+
+### 4. Unified Workflow
+
+#### 4.1 Single Input/Output
+
+```python
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={
+ "inputRvt": "path/to/input.rvt",
+ },
+ output_files={
+ "outputRvt": "output.rvt",
+ },
+ bucket_key="my-bucket",
+)
+```
+
+#### 4.2 Multiple Input/Output Files
+
+```python
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={
+ "inputRvt": "path/to/model.rvt",
+ "configJson": "path/to/config.json",
+ "templateRvt": "path/to/template.rvt",
+ },
+ output_files={
+ "outputRvt": "results/processed.rvt",
+ "reportPdf": "results/report.pdf",
+ "exportIfc": "results/export.ifc",
+ },
+ bucket_key="my-bucket",
+ download_outputs=True,
+ output_dir="./results",
+)
+```
+
+#### 4.3 Optional Input/Output Files
+
+Some Activities may not require input files or may only generate logs without output files:
+
+```python
+# No input files - Generate template from scratch
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.TemplateGenerator+prod",
+ output_files={
+ "outputRvt": "template.rvt",
+ },
+ bucket_key="my-bucket",
+)
+
+# No output files - Validation only (logs in report)
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.ValidationActivity+prod",
+ input_files={
+ "inputRvt": "model.rvt",
+ },
+ bucket_key="my-bucket",
+)
+
+# No files at all - Health check or scheduled task
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.HealthCheck+prod",
+)
+# Note: bucket_key not required when no files are involved
+
+# You can also pass empty dictionaries explicitly
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.MyActivity+prod",
+ input_files={},
+ output_files={},
+)
+```
+
+#### 4.4 Progress Monitoring
+
+```python
+import time
+
+def on_progress(status_data):
+ status = status_data.get("status")
+ progress = status_data.get("progress", "")
+ stats = status_data.get("stats", {})
+
+ print(f"\n[{time.strftime('%H:%M:%S')}] Status: {status}")
+
+ if progress:
+ print(f" Progress: {progress}")
+
+ if stats:
+ time_queued = stats.get("timeQueued")
+ time_download = stats.get("timeDownloadStarted")
+ time_processing = stats.get("timeInstructionsStarted")
+
+ if time_queued:
+ print(f" Queued at: {time_queued}")
+ if time_download:
+ print(f" Download started: {time_download}")
+ if time_processing:
+ print(f" Processing started: {time_processing}")
+
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={"inputRvt": "input.rvt"},
+ output_files={"outputRvt": "output.rvt"},
+ bucket_key="my-bucket",
+ on_progress=on_progress,
+)
+```
+
+---
+
+### 5. Batch Processing
+
+Execute multiple WorkItems concurrently:
+
+```python
+# Prepare batch WorkItem specifications
+workitems = []
+
+for i in range(1, 6):
+ input_url = workflow.upload_input_file(
+ local_path=f"inputs/model_{i}.rvt",
+ bucket_key="my-bucket",
+ object_key=f"batch/input_{i}.rvt",
+ )
+
+ output_url = workflow.prepare_output_url(
+ object_key=f"batch/output_{i}.rvt",
+ bucket_key="my-bucket",
+ )
+
+ workitems.append({
+ "activityId": "myowner.RevitActivity+prod",
+ "arguments": {
+ "inputRvt": {"url": input_url, "verb": "get"},
+ "outputRvt": {"url": output_url, "verb": "put"},
+ },
+ })
+
+# Execute batch
+results = workflow.run_batch_workitems(
+ workitems,
+ poll_interval=10.0,
+ timeout=3600.0,
+)
+
+# Process results
+for i, result in enumerate(results, 1):
+ print(f"WorkItem {i}: {result.status}")
+
+ if result.status == "success":
+ workflow.download_output_file(
+ bucket_key="my-bucket",
+ object_key=f"batch/output_{i}.rvt",
+ local_path=f"./results/output_{i}.rvt",
+ )
+```
+
+---
+
+## Webhook Callbacks
+
+Use webhooks instead of polling to receive WorkItem results asynchronously.
+
+### 1. Basic Usage
+
+```python
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={"inputRvt": "input.rvt"},
+ output_files={"outputRvt": "output.rvt"},
+ bucket_key="my-bucket",
+ # Webhook URL called on WorkItem completion
+ on_complete_url="https://myapp.com/api/webhooks/workitem-complete",
+ # Webhook URL called during WorkItem progress (optional)
+ on_progress_url="https://myapp.com/api/webhooks/workitem-progress",
+)
+```
+
+### 2. Webhook Server Implementation (Flask)
+
+```python
+from flask import Flask, request, jsonify
+
+app = Flask(__name__)
+
+@app.route('/api/webhooks/workitem-complete', methods=['POST'])
+def workitem_complete():
+ """
+ Called by Design Automation when WorkItem completes
+
+ Callback payload:
+ {
+ "id": "workitem-id",
+ "status": "success" | "failed" | "cancelled",
+ "reportUrl": "https://...",
+ "stats": {
+ "timeQueued": "2024-01-01T00:00:00Z",
+ "timeDownloadStarted": "2024-01-01T00:00:10Z",
+ "timeInstructionsStarted": "2024-01-01T00:00:20Z",
+ "timeInstructionsEnded": "2024-01-01T00:05:00Z",
+ "timeUploadEnded": "2024-01-01T00:05:30Z"
+ },
+ "activityId": "owner.ActivityName+alias",
+ ...
+ }
+ """
+ data = request.json
+
+ workitem_id = data.get('id')
+ status = data.get('status')
+ report_url = data.get('reportUrl')
+
+ print(f"WorkItem {workitem_id} completed with status: {status}")
+
+ if status == 'success':
+ # Success handling logic
+ # e.g., Update database, send notifications, download results
+ pass
+ elif status == 'failed':
+ # Failure handling logic
+ # e.g., Log errors, add to retry queue
+ pass
+
+ return jsonify({"received": True}), 200
+
+
+@app.route('/api/webhooks/workitem-progress', methods=['POST'])
+def workitem_progress():
+ """
+ Called by Design Automation during WorkItem execution
+
+ Callback payload:
+ {
+ "id": "workitem-id",
+ "status": "pending" | "inprogress",
+ "progress": "Downloading input files..." | "Processing..." | "Uploading results...",
+ ...
+ }
+ """
+ data = request.json
+
+ workitem_id = data.get('id')
+ status = data.get('status')
+ progress = data.get('progress', '')
+
+ print(f"WorkItem {workitem_id}: {status} - {progress}")
+
+ # Progress update logic
+ # e.g., WebSocket real-time notifications, database updates
+
+ return jsonify({"received": True}), 200
+
+
+if __name__ == '__main__':
+ # HTTPS required in production!
+ app.run(host='0.0.0.0', port=5000)
+```
+
+### 3. Public URL for Local Development
+
+Webhook URLs must be **publicly accessible HTTPS endpoints** that Design Automation can reach.
+
+#### Using ngrok
+
+```bash
+# 1. Install ngrok
+brew install ngrok # macOS
+# or download from https://ngrok.com/download
+
+# 2. Run ngrok
+ngrok http 5000
+
+# 3. Use generated URL
+# Forwarding: https://abc123.ngrok.io -> http://localhost:5000
+# → on_complete_url="https://abc123.ngrok.io/api/webhooks/workitem-complete"
+```
+
+#### Using Cloud Services
+
+- **AWS API Gateway + Lambda**
+- **Azure Functions**
+- **Google Cloud Functions**
+- **Vercel/Netlify Functions**
+
+### 4. Security Enhancement
+
+```python
+import hmac
+import hashlib
+from flask import Flask, request, jsonify, abort
+
+app = Flask(__name__)
+
+WEBHOOK_SECRET = "your-secret-key" # Recommended: use environment variable
+
+def verify_signature(payload: bytes, signature: str) -> bool:
+ """Verify webhook signature"""
+ expected = hmac.new(
+ WEBHOOK_SECRET.encode(),
+ payload,
+ hashlib.sha256
+ ).hexdigest()
+ return hmac.compare_digest(signature, expected)
+
+
+@app.route('/api/webhooks/workitem-complete', methods=['POST'])
+def secure_workitem_complete():
+ # 1. Signature verification (optional - Design Automation doesn't provide signatures by default)
+ # signature = request.headers.get('X-Webhook-Signature')
+ # if not verify_signature(request.data, signature):
+ # abort(401, "Invalid signature")
+
+ # 2. IP whitelist verification (optional)
+ # allowed_ips = ['52.x.x.x', '54.x.x.x'] # Autodesk IP ranges
+ # if request.remote_addr not in allowed_ips:
+ # abort(403, "Forbidden")
+
+ # 3. Process request data
+ data = request.json
+ workitem_id = data.get('id')
+
+ # 4. Ensure idempotency (prevent duplicate processing)
+ # if is_already_processed(workitem_id):
+ # return jsonify({"received": True, "note": "already processed"}), 200
+
+ # 5. Execute business logic
+ process_workitem_result(data)
+
+ return jsonify({"received": True}), 200
+```
+
+---
+
+## Advanced Usage
+
+### 1. Step-by-Step Workflow Control
+
+For fine-grained control over the entire process:
+
+```python
+# Step 1: Ensure bucket exists
+bucket = workflow.ensure_bucket("my-bucket")
+
+# Step 2: Upload input file
+input_url = workflow.upload_input_file(
+ local_path="path/to/input.rvt",
+ bucket_key="my-bucket",
+)
+
+# Step 3: Prepare output URL
+output_url = workflow.prepare_output_url(
+ object_key="output.rvt",
+ bucket_key="my-bucket",
+)
+
+# Step 4: Start WorkItem
+workitem_id = workflow.start_workitem(
+ activity_id="myowner.RevitActivity+prod",
+ arguments={
+ "inputRvt": {"url": input_url, "verb": "get"},
+ "outputRvt": {"url": output_url, "verb": "put"},
+ },
+)
+
+# Step 5: Wait for completion
+result = workflow.wait_for_completion(workitem_id)
+
+# Step 6: Download results
+if result.status == "success":
+ workflow.download_output_file(
+ bucket_key="my-bucket",
+ object_key="output.rvt",
+ local_path="./results/output.rvt",
+ )
+```
+
+### 2. Custom Timeout Settings
+
+```python
+# Increase timeout for long-running jobs
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={"inputRvt": "large-model.rvt"},
+ output_files={"outputRvt": "output.rvt"},
+ bucket_key="my-bucket",
+ poll_interval=30.0, # Check every 30 seconds
+ timeout=7200.0, # Wait up to 2 hours
+)
+```
+
+### 3. Selective Download
+
+```python
+# Disable automatic download (keep results in OSS only)
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={"inputRvt": "input.rvt"},
+ output_files={"outputRvt": "output.rvt"},
+ bucket_key="my-bucket",
+ download_outputs=False, # Don't download
+)
+
+# Download manually when needed
+if result.status == "success":
+ workflow.download_output_file(
+ bucket_key="my-bucket",
+ object_key="output.rvt",
+ local_path="./results/output.rvt",
+ )
+```
+
+---
+
+## Error Handling
+
+### 1. Basic Error Handling
+
+```python
+from pyaps.automation import AutomationError
+from pyaps.http import HTTPError
+
+try:
+ result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={"inputRvt": "input.rvt"},
+ output_files={"outputRvt": "output.rvt"},
+ bucket_key="my-bucket",
+ timeout=600.0,
+ )
+
+ if result.status == "success":
+ print("✓ WorkItem succeeded")
+ elif result.status == "failed":
+ print(f"✗ WorkItem failed")
+ print(f"Report URL: {result.report_url}")
+ if result.details:
+ print(f"Error details: {result.details}")
+
+except TimeoutError as e:
+ print(f"✗ Timeout: {e}")
+ # Can cancel WorkItem if needed
+ # workflow.cancel_workitem(workitem_id)
+
+except AutomationError as e:
+ print(f"✗ Automation error: {e}")
+ print(f"Status: {e.status}")
+ print(f"Payload: {e.payload}")
+
+except HTTPError as e:
+ print(f"✗ HTTP error: {e}")
+ print(f"Status: {e.status}")
+ print(f"Body: {e.body}")
+
+except ValueError as e:
+ print(f"✗ Configuration error: {e}")
+
+except Exception as e:
+ print(f"✗ Unexpected error: {e}")
+```
+
+### 2. Retry Logic
+
+```python
+import time
+
+def run_with_retry(workflow, max_retries=3):
+ for attempt in range(1, max_retries + 1):
+ try:
+ result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={"inputRvt": "input.rvt"},
+ output_files={"outputRvt": "output.rvt"},
+ bucket_key="my-bucket",
+ )
+
+ if result.status == "success":
+ return result
+ elif result.status == "failed":
+ print(f"Attempt {attempt} failed. Report: {result.report_url}")
+ if attempt < max_retries:
+ wait_time = 2 ** attempt # Exponential backoff
+ print(f"Retrying in {wait_time} seconds...")
+ time.sleep(wait_time)
+
+ except TimeoutError as e:
+ print(f"Attempt {attempt} timed out: {e}")
+ if attempt < max_retries:
+ time.sleep(60)
+
+ except Exception as e:
+ print(f"Attempt {attempt} error: {e}")
+ if attempt < max_retries:
+ time.sleep(30)
+
+ raise RuntimeError(f"Failed after {max_retries} attempts")
+
+# Usage
+result = run_with_retry(workflow)
+```
+
+---
+
+## Best Practices
+
+### 1. Use Environment Variables
+
+```python
+import os
+
+workflow = AutomationWorkflow(
+ automation_client=auto_client,
+ data_client=dm_client,
+ default_bucket=os.getenv("APS_DEFAULT_BUCKET", "my-default-bucket"),
+ poll_interval=float(os.getenv("APS_POLL_INTERVAL", "10.0")),
+ timeout=float(os.getenv("APS_TIMEOUT", "3600.0")),
+)
+```
+
+### 2. Choose Appropriate Bucket Policy
+
+- **transient** (24 hours): Testing and temporary work
+- **temporary** (30 days): Short-term projects
+- **persistent** (forever): Long-term storage needs
+
+```python
+workflow.ensure_bucket(
+ bucket_key="test-bucket",
+ policy_key="transient", # For testing
+)
+
+workflow.ensure_bucket(
+ bucket_key="production-bucket",
+ policy_key="persistent", # For production
+)
+```
+
+### 3. Enable Logging
+
+```python
+import logging
+
+logging.basicConfig(
+ level=logging.INFO,
+ format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+
+logger = logging.getLogger(__name__)
+
+def on_progress(status_data):
+ status = status_data.get("status")
+ progress = status_data.get("progress", "")
+ logger.info(f"WorkItem status: {status} - {progress}")
+
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={"inputRvt": "input.rvt"},
+ output_files={"outputRvt": "output.rvt"},
+ bucket_key="my-bucket",
+ on_progress=on_progress,
+)
+```
+
+### 4. Resource Cleanup
+
+```python
+from pyaps.datamanagement import DataManagementClient
+
+# Clean up temporary files
+def cleanup_temporary_files(dm_client: DataManagementClient, bucket_key: str):
+ """Delete temporary files older than 24 hours"""
+ import time
+
+ for obj in dm_client.buckets.list_objects(bucket_key):
+ object_key = obj.get("objectKey")
+ # Check creation time and delete if needed
+ # ...
+```
+
+### 5. Webhooks vs Polling
+
+| Method | Pros | Cons | Use Cases |
+|--------|------|------|-----------|
+| **Webhooks** | - Resource efficient
- Immediate notifications
- Serverless friendly | - Requires public endpoint
- More complex implementation | - Production environment
- Long-running jobs
- Multiple WorkItems |
+| **Polling** | - Simple implementation
- Easy local development | - Resource wasteful
- Polling delay | - Development/testing
- Single WorkItem
- Immediate results needed |
+
+```python
+# Development/Testing: Use polling
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={"inputRvt": "input.rvt"},
+ output_files={"outputRvt": "output.rvt"},
+ bucket_key="my-bucket",
+ poll_interval=10.0,
+)
+
+# Production: Use webhooks
+result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={"inputRvt": "input.rvt"},
+ output_files={"outputRvt": "output.rvt"},
+ bucket_key="my-bucket",
+ on_complete_url="https://myapp.com/webhooks/complete",
+)
+```
+
+---
+
+## API Reference
+
+### AutomationWorkflow
+
+#### Constructor
+
+```python
+AutomationWorkflow(
+ automation_client: AutomationClient,
+ data_client: DataManagementClient,
+ *,
+ default_bucket: Optional[str] = None,
+ poll_interval: float = 10.0,
+ timeout: float = 3600.0,
+)
+```
+
+#### Main Methods
+
+| Method | Description | Returns |
+|--------|-------------|---------|
+| `ensure_bucket(bucket_key, *, region, policy_key)` | Create or get existing bucket | `Dict[str, Any]` |
+| `upload_input_file(local_path, *, bucket_key, object_key, timeout)` | Upload input file | `str` (signed URL) |
+| `prepare_output_url(object_key, *, bucket_key, minutes_valid)` | Generate output URL | `str` (signed URL) |
+| `start_workitem(activity_id, arguments, *, nickname, on_complete, on_progress)` | Start WorkItem | `str` (workitem_id) |
+| `wait_for_completion(workitem_id, *, poll_interval, timeout, on_progress)` | Wait for completion | `WorkItemResult` |
+| `cancel_workitem(workitem_id)` | Cancel WorkItem | `None` |
+| `download_output_file(bucket_key, object_key, local_path)` | Download result | `None` |
+| `run_workitem_with_files(activity_id, input_files, output_files, ...)` | Unified workflow | `WorkItemResult` |
+| `run_batch_workitems(workitems, *, poll_interval, timeout)` | Batch processing | `List[WorkItemResult]` |
+
+### WorkItemResult
+
+```python
+@dataclass
+class WorkItemResult:
+ workitem_id: str
+ status: WorkItemStatus # "pending" | "inprogress" | "success" | "failed" | "cancelled"
+ report_url: Optional[str] = None
+ stats: Optional[Dict[str, Any]] = None
+ details: Optional[Dict[str, Any]] = None
+```
+
+---
+
+## Related Documentation
+
+- [AutomationClient API](./client.py) - Low-level API wrapper
+- [WorkItemSpec](./types.py) - WorkItem specification types
+- [Example Code](./workflow_example.py) - Runnable example code
+- [APS Design Automation API Docs](https://aps.autodesk.com/en/docs/design-automation/v3/)
+
+---
+
+## Troubleshooting
+
+### Q: "bucket_key must be provided" error
+
+```python
+# If default_bucket is not set, explicitly pass bucket_key
+workflow = AutomationWorkflow(
+ automation_client=auto_client,
+ data_client=dm_client,
+ default_bucket="my-bucket", # Add this
+)
+```
+
+### Q: TimeoutError occurs
+
+```python
+# Increase timeout or use webhooks
+result = workflow.run_workitem_with_files(
+ ...,
+ timeout=7200.0, # 2 hours
+ # Or use webhooks
+ on_complete_url="https://myapp.com/webhooks/complete",
+)
+```
+
+### Q: Webhook not being called
+
+1. **Check HTTPS**: HTTP is not supported
+2. **Public access**: localhost won't work (use ngrok, etc.)
+3. **Response code**: Ensure endpoint returns 200 OK
+4. **Timeout**: Webhook endpoint should respond quickly
+
+### Q: WorkItem status is "failed"
+
+```python
+# Check Report URL
+if result.status == "failed":
+ print(f"Report URL: {result.report_url}")
+ # Open in browser to see detailed error logs
+```
+
+---
+
+**Version**: 0.0.4
+**Last Updated**: 2025-01-30
diff --git a/src/pyaps/automation/__init__.py b/src/pyaps/automation/__init__.py
index 5908397..a845b8d 100644
--- a/src/pyaps/automation/__init__.py
+++ b/src/pyaps/automation/__init__.py
@@ -1,6 +1,7 @@
# src/pyaps/automation/__init__.py
from .client import AutomationClient, AutomationError, DEFAULT_AUTOMATION_SCOPES
from .types import WorkItemArgument, WorkItemSpec, AppBundleSpec, ActivitySpec
+from .workflow import AutomationWorkflow, WorkItemResult
__all__ = [
"AutomationClient",
@@ -10,4 +11,6 @@
"WorkItemSpec",
"AppBundleSpec",
"ActivitySpec",
+ "AutomationWorkflow",
+ "WorkItemResult",
]
diff --git a/src/pyaps/automation/types.py b/src/pyaps/automation/types.py
index 468c3c8..1451e8b 100644
--- a/src/pyaps/automation/types.py
+++ b/src/pyaps/automation/types.py
@@ -41,17 +41,27 @@ class WorkItemSpec:
WorkItem 생성 요청
- activity_id 예시: '{nickname}.{activity}+{alias}' 또는 '{owner}.{activity}+{alias}'
- arguments: Activity에서 선언한 파라미터 이름을 key로 사용
+ - on_complete: WorkItem 완료 시 호출될 콜백 URL (HTTP POST)
+ - on_progress: WorkItem 진행 상황 업데이트 시 호출될 콜백 URL (HTTP POST)
"""
activity_id: str
arguments: Dict[str, WorkItemArgument] = field(default_factory=dict)
nickname: Optional[str] = None
+ on_complete: Optional[str] = None
+ on_progress: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
- return {
+ d: Dict[str, Any] = {
"activityId": self.activity_id,
"arguments": {k: v.to_dict() for k, v in self.arguments.items()},
- **({"nickname": self.nickname} if self.nickname else {}),
}
+ if self.nickname:
+ d["nickname"] = self.nickname
+ if self.on_complete:
+ d["onComplete"] = self.on_complete
+ if self.on_progress:
+ d["onProgress"] = self.on_progress
+ return d
@dataclass
class AppBundleSpec:
diff --git a/src/pyaps/automation/workflow.py b/src/pyaps/automation/workflow.py
new file mode 100644
index 0000000..f377fa0
--- /dev/null
+++ b/src/pyaps/automation/workflow.py
@@ -0,0 +1,447 @@
+# src/pyaps/automation/workflow.py
+"""
+Design Automation 워크플로우 추상화
+OSS 버킷/오브젝트 준비 → WorkItem 실행 → 결과 다운로드까지의 전체 과정을 통합
+"""
+from __future__ import annotations
+
+import time
+from pathlib import Path
+from typing import Any, Callable, Dict, List, Literal, Optional, Tuple
+from dataclasses import dataclass
+
+from pyaps.automation.client import AutomationClient
+from pyaps.automation.types import WorkItemSpec, WorkItemArgument
+from pyaps.datamanagement.client import DataManagementClient
+
+WorkItemStatus = Literal["pending", "inprogress", "success", "failed", "cancelled"]
+
+
+@dataclass
+class WorkItemResult:
+ """WorkItem 실행 결과"""
+ workitem_id: str
+ status: WorkItemStatus
+ report_url: Optional[str] = None
+ stats: Optional[Dict[str, Any]] = None
+ details: Optional[Dict[str, Any]] = None
+
+
+class AutomationWorkflow:
+ """
+ Design Automation WorkItem 실행을 위한 고수준 워크플로우
+
+ 주요 기능:
+ 1. OSS 버킷/오브젝트 생성 및 파일 업로드
+ 2. WorkItem 실행 및 상태 모니터링
+ 3. 결과물 다운로드
+ """
+
+ def __init__(
+ self,
+ automation_client: AutomationClient,
+ data_client: DataManagementClient,
+ *,
+ default_bucket: Optional[str] = None,
+ poll_interval: float = 10.0,
+ timeout: float = 3600.0,
+ ):
+ """
+ Args:
+ automation_client: Design Automation API 클라이언트
+ data_client: Data Management API 클라이언트 (OSS 접근용)
+ default_bucket: 기본 OSS 버킷 키
+ poll_interval: WorkItem 상태 폴링 간격 (초)
+ timeout: WorkItem 최대 대기 시간 (초)
+ """
+ self.auto = automation_client
+ self.dm = data_client
+ self.default_bucket = default_bucket
+ self.poll_interval = poll_interval
+ self.timeout = timeout
+
+ # ==================== Step 1: OSS 준비 ====================
+
+ def ensure_bucket(
+ self,
+ bucket_key: str,
+ *,
+ region: str = "US",
+ policy_key: str = "transient",
+ ) -> Dict[str, Any]:
+ """
+ OSS 버킷 생성 (이미 존재하면 무시)
+
+ Args:
+ bucket_key: 버킷 키 (소문자, 숫자, 하이픈만 허용, 3-128자)
+ region: 리전 (US, EMEA 등)
+ policy_key: 보관 정책 (transient=24시간, temporary=30일, persistent=영구)
+
+ Returns:
+ 버킷 정보
+ """
+ try:
+ return self.dm.buckets.get(bucket_key)
+ except Exception:
+ # 버킷이 없으면 생성
+ return self.dm.buckets.create(
+ bucket_key,
+ region=region,
+ policy_key=policy_key,
+ )
+
+ def upload_input_file(
+ self,
+ local_path: str | Path,
+ *,
+ bucket_key: Optional[str] = None,
+ object_key: Optional[str] = None,
+ timeout: Optional[float] = None,
+ ) -> str:
+ """
+ 입력 파일을 OSS에 업로드하고 다운로드 URL 반환
+
+ Args:
+ local_path: 로컬 파일 경로
+ bucket_key: OSS 버킷 키 (미지정시 default_bucket 사용)
+ object_key: OSS 오브젝트 키 (미지정시 파일명 사용)
+ timeout: 업로드 타임아웃
+
+ Returns:
+ Signed download URL (WorkItem arguments에서 사용)
+ """
+ bucket_key = bucket_key or self.default_bucket
+ if not bucket_key:
+ raise ValueError("bucket_key must be provided or default_bucket must be set")
+
+ local_path = Path(local_path)
+ object_key = object_key or local_path.name
+
+ # OSS에 업로드
+ signed_upload = self.dm.objects.post_signed(
+ bucket_key, object_key, access="readwrite"
+ )
+
+ with open(local_path, "rb") as f:
+ file_data = f.read()
+
+ self.dm.objects.upload_via_signed(signed_upload, file_data, timeout=timeout)
+
+ # 다운로드 URL 생성 (WorkItem에서 사용할 것)
+ signed_download = self.dm.objects.get_signed_download(
+ bucket_key, object_key, minutes_valid=60
+ )
+ return signed_download.get("url") or signed_download.get("signedUrl")
+
+ def prepare_output_url(
+ self,
+ object_key: str,
+ *,
+ bucket_key: Optional[str] = None,
+ minutes_valid: int = 60,
+ ) -> str:
+ """
+ 출력 파일용 업로드 URL 생성
+
+ Args:
+ object_key: OSS 오브젝트 키
+ bucket_key: OSS 버킷 키 (미지정시 default_bucket 사용)
+ minutes_valid: URL 유효 시간 (분)
+
+ Returns:
+ Signed upload URL (WorkItem arguments에서 사용)
+ """
+ bucket_key = bucket_key or self.default_bucket
+ if not bucket_key:
+ raise ValueError("bucket_key must be provided or default_bucket must be set")
+
+ # 업로드용 signed URL 생성
+ signed = self.dm.objects.post_signed(
+ bucket_key, object_key, access="readwrite"
+ )
+ return signed.get("url") or signed.get("signedUrl")
+
+ # ==================== Step 2: WorkItem 실행 ====================
+
+ def start_workitem(
+ self,
+ activity_id: str,
+ arguments: Dict[str, WorkItemArgument | Dict[str, Any]],
+ *,
+ nickname: Optional[str] = None,
+ on_complete: Optional[str] = None,
+ on_progress: Optional[str] = None,
+ ) -> str:
+ """
+ WorkItem 시작
+
+ Args:
+ activity_id: Activity 전체 ID (예: 'owner.ActivityName+alias')
+ arguments: Activity 파라미터별 입출력 URL
+ nickname: WorkItem 별칭 (선택)
+ on_complete: WorkItem 완료 시 호출될 콜백 URL (HTTP POST로 WorkItem 정보 전송)
+ on_progress: WorkItem 진행 상황 업데이트 시 호출될 콜백 URL (HTTP POST)
+
+ Returns:
+ WorkItem ID
+
+ Note:
+ 콜백 URL은 공개적으로 접근 가능한 HTTPS 엔드포인트여야 합니다.
+ onComplete 콜백에는 WorkItem 상태, 결과, 통계 등이 포함된 JSON이 전송됩니다.
+ """
+ # WorkItemArgument 변환
+ converted_args = {}
+ for key, arg in arguments.items():
+ if isinstance(arg, dict):
+ converted_args[key] = WorkItemArgument(**arg)
+ else:
+ converted_args[key] = arg
+
+ spec = WorkItemSpec(
+ activity_id=activity_id,
+ arguments=converted_args,
+ nickname=nickname,
+ on_complete=on_complete,
+ on_progress=on_progress,
+ )
+
+ result = self.auto.start_workitem(spec)
+ return result["id"]
+
+ def wait_for_completion(
+ self,
+ workitem_id: str,
+ *,
+ poll_interval: Optional[float] = None,
+ timeout: Optional[float] = None,
+ on_progress: Optional[Callable[[Dict[str, Any]], None]] = None,
+ ) -> WorkItemResult:
+ """
+ WorkItem 완료 대기
+
+ Args:
+ workitem_id: WorkItem ID
+ poll_interval: 폴링 간격 (초)
+ timeout: 최대 대기 시간 (초)
+ on_progress: 진행 상황 콜백 함수
+
+ Returns:
+ WorkItem 실행 결과
+
+ Raises:
+ TimeoutError: 타임아웃 발생
+ RuntimeError: WorkItem 실행 실패
+ """
+ poll_interval = poll_interval or self.poll_interval
+ timeout = timeout or self.timeout
+
+ start_time = time.time()
+
+ while True:
+ elapsed = time.time() - start_time
+ if elapsed > timeout:
+ raise TimeoutError(
+ f"WorkItem {workitem_id} timed out after {timeout}s"
+ )
+
+ status_data = self.auto.get_workitem(workitem_id)
+ status = status_data.get("status")
+
+ if on_progress:
+ on_progress(status_data)
+
+ if status in ("success", "failed", "cancelled"):
+ return WorkItemResult(
+ workitem_id=workitem_id,
+ status=status,
+ report_url=status_data.get("reportUrl"),
+ stats=status_data.get("stats"),
+ details=status_data,
+ )
+
+ time.sleep(poll_interval)
+
+ def cancel_workitem(self, workitem_id: str) -> None:
+ """WorkItem 취소"""
+ self.auto.cancel_workitem(workitem_id)
+
+ # ==================== Step 3: 결과 다운로드 ====================
+
+ def download_output_file(
+ self,
+ bucket_key: str,
+ object_key: str,
+ local_path: str | Path,
+ ) -> None:
+ """
+ OSS에서 출력 파일 다운로드
+
+ Args:
+ bucket_key: OSS 버킷 키
+ object_key: OSS 오브젝트 키
+ local_path: 저장할 로컬 경로
+ """
+ # Signed download URL 생성
+ signed = self.dm.objects.get_signed_download(
+ bucket_key, object_key, minutes_valid=10
+ )
+ url = signed.get("url") or signed.get("signedUrl")
+
+ # 파일 다운로드
+ import requests
+ response = requests.get(url, timeout=300)
+ response.raise_for_status()
+
+ local_path = Path(local_path)
+ local_path.parent.mkdir(parents=True, exist_ok=True)
+
+ with open(local_path, "wb") as f:
+ f.write(response.content)
+
+ # ==================== 통합 워크플로우 ====================
+
+ def run_workitem_with_files(
+ self,
+ activity_id: str,
+ input_files: Optional[Dict[str, str | Path]] = None,
+ output_files: Optional[Dict[str, str]] = None,
+ *,
+ bucket_key: Optional[str] = None,
+ download_outputs: bool = True,
+ output_dir: Optional[str | Path] = None,
+ poll_interval: Optional[float] = None,
+ timeout: Optional[float] = None,
+ on_progress: Optional[Callable[[Dict[str, Any]], None]] = None,
+ on_complete_url: Optional[str] = None,
+ on_progress_url: Optional[str] = None,
+ ) -> WorkItemResult:
+ """
+ 파일 업로드 → WorkItem 실행 → 결과 다운로드 전체 워크플로우
+
+ Args:
+ activity_id: Activity 전체 ID
+ input_files: 입력 파일 매핑 {argument_name: local_file_path} (선택, None 또는 {} 가능)
+ output_files: 출력 파일 매핑 {argument_name: object_key} (선택, None 또는 {} 가능)
+ bucket_key: OSS 버킷 키 (입력/출력 파일이 있을 때만 필요)
+ download_outputs: 완료 후 출력 파일 자동 다운로드 여부
+ output_dir: 출력 파일 저장 디렉토리
+ poll_interval: 폴링 간격 (초)
+ timeout: 최대 대기 시간 (초)
+ on_progress: 진행 상황 콜백 함수 (로컬 폴링용)
+ on_complete_url: WorkItem 완료 시 호출될 웹훅 URL (Design Automation에서 HTTP POST)
+ on_progress_url: WorkItem 진행 중 호출될 웹훅 URL (Design Automation에서 HTTP POST)
+
+ Returns:
+ WorkItem 실행 결과
+
+ Example:
+ >>> # 일반적인 사용
+ >>> result = workflow.run_workitem_with_files(
+ ... activity_id="myowner.RevitActivity+prod",
+ ... input_files={"inputRvt": "input.rvt"},
+ ... output_files={"outputRvt": "output.rvt"},
+ ... bucket_key="my-bucket",
+ ... )
+ >>>
+ >>> # 입력 없이 출력만 생성 (예: 템플릿 생성)
+ >>> result = workflow.run_workitem_with_files(
+ ... activity_id="myowner.TemplateGenerator+prod",
+ ... output_files={"outputRvt": "template.rvt"},
+ ... bucket_key="my-bucket",
+ ... )
+ >>>
+ >>> # 로그만 생성 (입출력 파일 없음)
+ >>> result = workflow.run_workitem_with_files(
+ ... activity_id="myowner.ValidationActivity+prod",
+ ... )
+
+ Note:
+ on_complete_url을 사용하면 폴링 없이 비동기로 결과를 받을 수 있습니다.
+ 콜백 URL은 공개적으로 접근 가능한 HTTPS 엔드포인트여야 합니다.
+ """
+ input_files = input_files or {}
+ output_files = output_files or {}
+
+ # 파일이 있는 경우에만 버킷 확인
+ if input_files or output_files:
+ bucket_key = bucket_key or self.default_bucket
+ if not bucket_key:
+ raise ValueError(
+ "bucket_key must be provided or default_bucket must be set when using input_files or output_files"
+ )
+ # 버킷 확인/생성
+ self.ensure_bucket(bucket_key)
+
+ # 1. 입력 파일 업로드
+ arguments: Dict[str, WorkItemArgument] = {}
+
+ for arg_name, local_path in input_files.items():
+ url = self.upload_input_file(local_path, bucket_key=bucket_key)
+ arguments[arg_name] = WorkItemArgument(url=url, verb="get")
+
+ # 2. 출력 URL 준비
+ for arg_name, object_key in output_files.items():
+ url = self.prepare_output_url(object_key, bucket_key=bucket_key)
+ arguments[arg_name] = WorkItemArgument(url=url, verb="put")
+
+ # 3. WorkItem 실행 (콜백 URL 포함)
+ workitem_id = self.start_workitem(
+ activity_id,
+ arguments,
+ on_complete=on_complete_url,
+ on_progress=on_progress_url,
+ )
+
+ # 4. 완료 대기
+ result = self.wait_for_completion(
+ workitem_id,
+ poll_interval=poll_interval,
+ timeout=timeout,
+ on_progress=on_progress,
+ )
+
+ # 5. 결과 다운로드
+ if download_outputs and result.status == "success" and output_files:
+ output_dir = Path(output_dir) if output_dir else Path.cwd()
+
+ for arg_name, object_key in output_files.items():
+ local_path = output_dir / object_key
+ self.download_output_file(bucket_key, object_key, local_path)
+
+ return result
+
+ # ==================== 배치 처리 ====================
+
+ def run_batch_workitems(
+ self,
+ workitems: List[Dict[str, Any]],
+ *,
+ poll_interval: Optional[float] = None,
+ timeout: Optional[float] = None,
+ ) -> List[WorkItemResult]:
+ """
+ 여러 WorkItem을 배치로 실행하고 모두 완료될 때까지 대기
+
+ Args:
+ workitems: WorkItem 스펙 목록
+ poll_interval: 폴링 간격 (초)
+ timeout: 최대 대기 시간 (초)
+
+ Returns:
+ 각 WorkItem의 실행 결과 목록
+ """
+ # 배치 시작
+ batch_result = self.auto.create_workitems_batch(workitems)
+ workitem_ids = [wi["id"] for wi in batch_result]
+
+ # 모든 WorkItem 완료 대기
+ results = []
+ for workitem_id in workitem_ids:
+ result = self.wait_for_completion(
+ workitem_id,
+ poll_interval=poll_interval,
+ timeout=timeout,
+ )
+ results.append(result)
+
+ return results
diff --git a/src/pyaps/automation/workflow_example.py b/src/pyaps/automation/workflow_example.py
new file mode 100644
index 0000000..2b30b70
--- /dev/null
+++ b/src/pyaps/automation/workflow_example.py
@@ -0,0 +1,595 @@
+"""
+Design Automation Workflow 사용 예제
+파일 업로드 → WorkItem 실행 → 결과 다운로드까지의 전체 워크플로우
+"""
+from __future__ import annotations
+
+import os
+import time
+from pathlib import Path
+
+from pyaps.auth import AuthClient, InMemoryTokenStore
+from pyaps.automation import (
+ AutomationClient,
+ AutomationWorkflow,
+ DEFAULT_AUTOMATION_SCOPES,
+)
+from pyaps.datamanagement import DataManagementClient
+
+
+# Load .env file if exists
+def load_dotenv():
+ """Simple .env loader"""
+ env_file = Path(__file__).parent.parent.parent.parent / ".env"
+ if env_file.exists():
+ with open(env_file) as f:
+ for line in f:
+ line = line.strip()
+ if line and not line.startswith("#") and "=" in line:
+ key, value = line.split("=", 1)
+ os.environ[key.strip()] = value.strip()
+
+
+load_dotenv()
+
+
+# Environment variables
+APS_CLIENT_ID = os.getenv("APS_CLIENT_ID")
+APS_CLIENT_SECRET = os.getenv("APS_CLIENT_SECRET")
+APS_REGION = os.getenv("APS_REGION") or "us-east"
+
+
+def create_workflow() -> AutomationWorkflow:
+ """Create AutomationWorkflow instance"""
+ auth_client = AuthClient(
+ client_id=APS_CLIENT_ID,
+ client_secret=APS_CLIENT_SECRET,
+ store=InMemoryTokenStore(),
+ )
+
+ def token_provider() -> str:
+ token = auth_client.two_legged.get_token(DEFAULT_AUTOMATION_SCOPES)
+ return token.access_token
+
+ auto = AutomationClient(
+ token_provider=token_provider,
+ region=APS_REGION,
+ user_agent="pyaps-automation-workflow",
+ timeout=30.0,
+ )
+
+ dm = DataManagementClient(
+ token_provider=token_provider,
+ user_agent="pyaps-automation-workflow",
+ timeout=30.0,
+ )
+
+ return AutomationWorkflow(
+ automation_client=auto,
+ data_client=dm,
+ default_bucket="my-design-automation-bucket", # 기본 버킷 설정
+ poll_interval=10.0, # 10초마다 상태 확인
+ timeout=3600.0, # 최대 1시간 대기
+ )
+
+
+def example_simple_workflow():
+ """
+ 가장 간단한 워크플로우 예제:
+ 로컬 파일 → OSS 업로드 → WorkItem 실행 → 결과 다운로드
+ """
+ print("\n" + "=" * 60)
+ print("Example 1: Simple Workflow")
+ print("=" * 60)
+
+ workflow = create_workflow()
+
+ # 전체 워크플로우를 한 번에 실행
+ result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={
+ "inputRvt": "path/to/input.rvt", # 로컬 파일
+ },
+ output_files={
+ "outputRvt": "output.rvt", # OSS 오브젝트 키
+ },
+ bucket_key="my-bucket",
+ download_outputs=True, # 완료 후 자동 다운로드
+ output_dir="./results", # 결과 저장 위치
+ )
+
+ print(f"WorkItem ID: {result.workitem_id}")
+ print(f"Status: {result.status}")
+ print(f"Report URL: {result.report_url}")
+
+
+def example_step_by_step_workflow():
+ """
+ 단계별 워크플로우 예제:
+ 각 단계를 개별적으로 실행
+ """
+ print("\n" + "=" * 60)
+ print("Example 2: Step-by-Step Workflow")
+ print("=" * 60)
+
+ workflow = create_workflow()
+
+ # Step 1: 버킷 확인/생성
+ print("\n[Step 1] Ensure bucket exists")
+ bucket = workflow.ensure_bucket(
+ bucket_key="my-design-automation-bucket",
+ region="US",
+ policy_key="transient", # 24시간 보관
+ )
+ print(f"✓ Bucket ready: {bucket.get('bucketKey')}")
+
+ # Step 2: 입력 파일 업로드
+ print("\n[Step 2] Upload input file")
+ input_url = workflow.upload_input_file(
+ local_path="path/to/input.rvt",
+ bucket_key="my-design-automation-bucket",
+ object_key="inputs/input.rvt",
+ )
+ print(f"✓ Input uploaded: {input_url[:50]}...")
+
+ # Step 3: 출력 URL 준비
+ print("\n[Step 3] Prepare output URL")
+ output_url = workflow.prepare_output_url(
+ object_key="outputs/output.rvt",
+ bucket_key="my-design-automation-bucket",
+ )
+ print(f"✓ Output URL ready: {output_url[:50]}...")
+
+ # Step 4: WorkItem 시작
+ print("\n[Step 4] Start WorkItem")
+ workitem_id = workflow.start_workitem(
+ activity_id="myowner.RevitActivity+prod",
+ arguments={
+ "inputRvt": {"url": input_url, "verb": "get"},
+ "outputRvt": {"url": output_url, "verb": "put"},
+ },
+ )
+ print(f"✓ WorkItem started: {workitem_id}")
+
+ # Step 5: 완료 대기
+ print("\n[Step 5] Wait for completion")
+
+ def on_progress(status_data):
+ status = status_data.get("status")
+ progress = status_data.get("progress", "")
+ print(f" Status: {status} {progress}")
+
+ result = workflow.wait_for_completion(
+ workitem_id,
+ poll_interval=10.0,
+ timeout=3600.0,
+ on_progress=on_progress,
+ )
+ print(f"✓ Completed: {result.status}")
+
+ # Step 6: 결과 다운로드
+ if result.status == "success":
+ print("\n[Step 6] Download output")
+ workflow.download_output_file(
+ bucket_key="my-design-automation-bucket",
+ object_key="outputs/output.rvt",
+ local_path="./results/output.rvt",
+ )
+ print("✓ Output downloaded")
+
+
+def example_multiple_files():
+ """
+ 여러 파일을 입출력으로 사용하는 예제
+ """
+ print("\n" + "=" * 60)
+ print("Example 3: Multiple Input/Output Files")
+ print("=" * 60)
+
+ workflow = create_workflow()
+
+ result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={
+ "inputRvt": "path/to/model.rvt",
+ "configJson": "path/to/config.json",
+ "templateRvt": "path/to/template.rvt",
+ },
+ output_files={
+ "outputRvt": "results/processed.rvt",
+ "reportPdf": "results/report.pdf",
+ "exportIfc": "results/export.ifc",
+ },
+ bucket_key="my-bucket",
+ download_outputs=True,
+ output_dir="./results",
+ )
+
+ print(f"Status: {result.status}")
+
+
+def example_batch_processing():
+ """
+ 여러 WorkItem을 배치로 실행하는 예제
+ """
+ print("\n" + "=" * 60)
+ print("Example 4: Batch Processing")
+ print("=" * 60)
+
+ workflow = create_workflow()
+
+ # 여러 파일을 처리할 WorkItem 스펙 준비
+ workitems = []
+
+ for i in range(1, 6):
+ # 각 파일별로 입력/출력 URL 준비
+ input_url = workflow.upload_input_file(
+ local_path=f"inputs/model_{i}.rvt",
+ bucket_key="my-bucket",
+ object_key=f"batch/input_{i}.rvt",
+ )
+
+ output_url = workflow.prepare_output_url(
+ object_key=f"batch/output_{i}.rvt",
+ bucket_key="my-bucket",
+ )
+
+ workitems.append(
+ {
+ "activityId": "myowner.RevitActivity+prod",
+ "arguments": {
+ "inputRvt": {"url": input_url, "verb": "get"},
+ "outputRvt": {"url": output_url, "verb": "put"},
+ },
+ }
+ )
+
+ # 배치 실행
+ results = workflow.run_batch_workitems(
+ workitems,
+ poll_interval=10.0,
+ timeout=3600.0,
+ )
+
+ # 결과 확인
+ for i, result in enumerate(results, 1):
+ print(f"WorkItem {i}: {result.status}")
+
+ if result.status == "success":
+ workflow.download_output_file(
+ bucket_key="my-bucket",
+ object_key=f"batch/output_{i}.rvt",
+ local_path=f"./results/output_{i}.rvt",
+ )
+
+
+def example_error_handling():
+ """
+ 에러 처리 예제
+ """
+ print("\n" + "=" * 60)
+ print("Example 5: Error Handling")
+ print("=" * 60)
+
+ workflow = create_workflow()
+
+ try:
+ result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={"inputRvt": "path/to/input.rvt"},
+ output_files={"outputRvt": "output.rvt"},
+ bucket_key="my-bucket",
+ timeout=600.0, # 10분 타임아웃
+ )
+
+ if result.status == "success":
+ print("✓ WorkItem succeeded")
+ elif result.status == "failed":
+ print(f"✗ WorkItem failed")
+ print(f"Report URL: {result.report_url}")
+ if result.details:
+ print(f"Error details: {result.details}")
+
+ except TimeoutError as e:
+ print(f"✗ Timeout: {e}")
+ # WorkItem 취소 가능
+ # workflow.cancel_workitem(workitem_id)
+
+ except Exception as e:
+ print(f"✗ Error: {e}")
+
+
+def example_webhook_callbacks():
+ """
+ 웹훅 콜백 사용 예제 (onComplete, onProgress)
+ """
+ print("\n" + "=" * 60)
+ print("Example 6: Webhook Callbacks")
+ print("=" * 60)
+
+ workflow = create_workflow()
+
+ # 콜백 URL 사용 시 폴링 없이 비동기로 실행 가능
+ result = workflow.run_workitem_with_files(
+ activity_id="myowner.RevitActivity+prod",
+ input_files={"inputRvt": "path/to/input.rvt"},
+ output_files={"outputRvt": "output.rvt"},
+ bucket_key="my-bucket",
+ # WorkItem 완료 시 호출될 웹훅 URL
+ on_complete_url="https://myapp.com/api/webhooks/workitem-complete",
+ # WorkItem 진행 중 호출될 웹훅 URL (선택)
+ on_progress_url="https://myapp.com/api/webhooks/workitem-progress",
+ )
+
+ print(f"✓ WorkItem started: {result.workitem_id}")
+ print(f" Complete callback will be sent to: https://myapp.com/api/webhooks/workitem-complete")
+
+
+def example_webhook_callback_server():
+ """
+ 웹훅 콜백을 받는 서버 예제 (Flask)
+ """
+ print("\n" + "=" * 60)
+ print("Example 7: Webhook Callback Server (Flask)")
+ print("=" * 60)
+
+ print("""
+# Flask 서버 예제 - 콜백을 받는 엔드포인트
+
+from flask import Flask, request, jsonify
+
+app = Flask(__name__)
+
+@app.route('/api/webhooks/workitem-complete', methods=['POST'])
+def workitem_complete():
+ '''
+ Design Automation에서 WorkItem 완료 시 호출됨
+
+ Callback payload 구조:
+ {
+ "id": "workitem-id",
+ "status": "success" | "failed" | "cancelled",
+ "reportUrl": "https://...",
+ "stats": {
+ "timeQueued": "2024-01-01T00:00:00Z",
+ "timeDownloadStarted": "2024-01-01T00:00:10Z",
+ "timeInstructionsStarted": "2024-01-01T00:00:20Z",
+ "timeInstructionsEnded": "2024-01-01T00:05:00Z",
+ "timeUploadEnded": "2024-01-01T00:05:30Z"
+ },
+ "activityId": "owner.ActivityName+alias",
+ ...
+ }
+ '''
+ data = request.json
+
+ workitem_id = data.get('id')
+ status = data.get('status')
+ report_url = data.get('reportUrl')
+
+ print(f"WorkItem {workitem_id} completed with status: {status}")
+
+ if status == 'success':
+ # 성공 처리 로직
+ print(f" Success! Report: {report_url}")
+ # 예: 데이터베이스 업데이트, 알림 전송 등
+
+ elif status == 'failed':
+ # 실패 처리 로직
+ print(f" Failed! Report: {report_url}")
+ # 예: 에러 로깅, 재시도 큐에 추가 등
+
+ return jsonify({"received": True}), 200
+
+
+@app.route('/api/webhooks/workitem-progress', methods=['POST'])
+def workitem_progress():
+ '''
+ Design Automation에서 WorkItem 진행 중 호출됨
+
+ Callback payload 구조:
+ {
+ "id": "workitem-id",
+ "status": "pending" | "inprogress",
+ "progress": "Downloading input files..." | "Processing..." | "Uploading results...",
+ ...
+ }
+ '''
+ data = request.json
+
+ workitem_id = data.get('id')
+ status = data.get('status')
+ progress = data.get('progress', '')
+
+ print(f"WorkItem {workitem_id}: {status} - {progress}")
+
+ # 진행 상황 업데이트 로직 (예: WebSocket으로 실시간 알림)
+
+ return jsonify({"received": True}), 200
+
+
+if __name__ == '__main__':
+ # 프로덕션에서는 HTTPS 필수!
+ # ngrok, AWS API Gateway, Azure Functions 등 사용 권장
+ app.run(host='0.0.0.0', port=5000, ssl_context='adhoc')
+ """)
+
+ print("\n💡 로컬 개발 시 공개 URL 생성 방법:")
+ print(" 1. ngrok 사용:")
+ print(" $ ngrok http 5000")
+ print(" → https://abc123.ngrok.io → 이 URL을 on_complete_url에 사용")
+ print()
+ print(" 2. 클라우드 서비스 사용:")
+ print(" - AWS API Gateway + Lambda")
+ print(" - Azure Functions")
+ print(" - Google Cloud Functions")
+ print(" - Vercel/Netlify Functions")
+
+
+def example_webhook_with_signature():
+ """
+ 보안 강화: 서명 검증을 포함한 웹훅 처리 예제
+ """
+ print("\n" + "=" * 60)
+ print("Example 8: Secure Webhook with Signature Verification")
+ print("=" * 60)
+
+ print("""
+# 서명 검증을 포함한 보안 강화 예제
+
+import hmac
+import hashlib
+from flask import Flask, request, jsonify, abort
+
+app = Flask(__name__)
+
+# 웹훅 비밀키 (환경 변수로 관리 권장)
+WEBHOOK_SECRET = "your-secret-key"
+
+
+def verify_signature(payload: bytes, signature: str) -> bool:
+ '''서명 검증'''
+ expected = hmac.new(
+ WEBHOOK_SECRET.encode(),
+ payload,
+ hashlib.sha256
+ ).hexdigest()
+ return hmac.compare_digest(signature, expected)
+
+
+@app.route('/api/webhooks/workitem-complete', methods=['POST'])
+def secure_workitem_complete():
+ # 1. 서명 검증 (선택사항 - Design Automation은 기본적으로 서명 제공 안 함)
+ # signature = request.headers.get('X-Webhook-Signature')
+ # if not verify_signature(request.data, signature):
+ # abort(401, "Invalid signature")
+
+ # 2. IP 화이트리스트 검증 (선택사항)
+ # allowed_ips = ['52.x.x.x', '54.x.x.x'] # Autodesk IP 범위
+ # if request.remote_addr not in allowed_ips:
+ # abort(403, "Forbidden")
+
+ # 3. 요청 데이터 처리
+ data = request.json
+ workitem_id = data.get('id')
+
+ # 4. 멱등성 보장 (중복 요청 방지)
+ # if is_already_processed(workitem_id):
+ # return jsonify({"received": True, "note": "already processed"}), 200
+
+ # 5. 비즈니스 로직 실행
+ process_workitem_result(data)
+
+ return jsonify({"received": True}), 200
+
+
+def process_workitem_result(data):
+ '''WorkItem 결과 처리'''
+ workitem_id = data.get('id')
+ status = data.get('status')
+
+ # 비동기 처리 권장 (Celery, RQ 등)
+ # task_queue.enqueue(process_result, workitem_id, status)
+
+ print(f"Processing WorkItem {workitem_id}: {status}")
+ """)
+
+
+def example_progress_monitoring():
+ """
+ 진행 상황 모니터링 예제
+ """
+ print("\n" + "=" * 60)
+ print("Example 6: Progress Monitoring")
+ print("=" * 60)
+
+ workflow = create_workflow()
+
+ # 입력/출력 URL 준비
+ input_url = workflow.upload_input_file(
+ local_path="path/to/input.rvt",
+ bucket_key="my-bucket",
+ )
+
+ output_url = workflow.prepare_output_url(
+ object_key="output.rvt",
+ bucket_key="my-bucket",
+ )
+
+ # WorkItem 시작
+ workitem_id = workflow.start_workitem(
+ activity_id="myowner.RevitActivity+prod",
+ arguments={
+ "inputRvt": {"url": input_url, "verb": "get"},
+ "outputRvt": {"url": output_url, "verb": "put"},
+ },
+ )
+
+ # 상세한 진행 상황 모니터링
+ def on_progress(status_data):
+ status = status_data.get("status")
+ progress = status_data.get("progress", "")
+ stats = status_data.get("stats", {})
+
+ print(f"\n[{time.strftime('%H:%M:%S')}] Status: {status}")
+
+ if progress:
+ print(f" Progress: {progress}")
+
+ if stats:
+ time_queued = stats.get("timeQueued")
+ time_download = stats.get("timeDownloadStarted")
+ time_instr = stats.get("timeInstructionsStarted")
+ time_upload = stats.get("timeUploadEnded")
+
+ if time_queued:
+ print(f" Queued at: {time_queued}")
+ if time_download:
+ print(f" Download started: {time_download}")
+ if time_instr:
+ print(f" Processing started: {time_instr}")
+ if time_upload:
+ print(f" Upload ended: {time_upload}")
+
+ result = workflow.wait_for_completion(
+ workitem_id,
+ on_progress=on_progress,
+ )
+
+ print(f"\n✓ Final status: {result.status}")
+ if result.report_url:
+ print(f" Report: {result.report_url}")
+
+
+def main():
+ """예제 실행"""
+ print("\n" + "=" * 60)
+ print("Design Automation Workflow Examples")
+ print("=" * 60)
+
+ if not APS_CLIENT_ID or not APS_CLIENT_SECRET:
+ print("\n⚠ Set environment variables:")
+ print(" export APS_CLIENT_ID='your_client_id'")
+ print(" export APS_CLIENT_SECRET='your_client_secret'")
+ print(" export APS_REGION='us-east' # or 'eu-west'")
+ return
+
+ print("\n📚 Available examples:")
+ print(" 1. Simple Workflow - 한 줄로 전체 프로세스 실행")
+ print(" 2. Step-by-Step - 각 단계를 개별적으로 실행")
+ print(" 3. Multiple Files - 여러 입출력 파일 처리")
+ print(" 4. Batch Processing - 여러 WorkItem 배치 실행")
+ print(" 5. Error Handling - 에러 처리")
+ print(" 6. Webhook Callbacks - onComplete/onProgress 웹훅 사용")
+ print(" 7. Webhook Server - Flask 웹훅 서버 예제")
+ print(" 8. Secure Webhook - 보안 강화 웹훅 처리")
+ print(" 9. Progress Monitoring - 진행 상황 모니터링")
+
+ print("\n💡 Usage:")
+ print(" from pyaps.automation import AutomationWorkflow")
+ print(" workflow = AutomationWorkflow(auto_client, dm_client)")
+ print(" result = workflow.run_workitem_with_files(...)")
+
+
+if __name__ == "__main__":
+ main()