From 0d7be2e73bb825eb807625555831b566f5c0b4aa Mon Sep 17 00:00:00 2001 From: hjnoh Date: Thu, 30 Oct 2025 08:49:13 +0900 Subject: [PATCH] FEAT: Add AutomationWorkflow high-level API for Design Automation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add comprehensive workflow abstraction for Design Automation WorkItem execution: - New AutomationWorkflow class for unified workflow management * OSS bucket/object preparation * File upload/download with automatic URL generation * WorkItem execution and status monitoring * Batch processing support * Optional input/output files support - Webhook callback support (onComplete/onProgress) * Add callback URL parameters to WorkItemSpec * Support async notifications via webhooks * No polling required when using webhooks - Comprehensive documentation * WORKFLOW.md: Complete user guide with examples * workflow_example.py: 8+ runnable examples * Webhook server implementation examples (Flask) * Security best practices - Flexible file handling * Optional input_files and output_files parameters * Support Activities without input/output files * Smart bucket validation (only when files are present) - API improvements * Selective download control * Custom timeout settings * Progress monitoring callbacks * Error handling and retry patterns ๐Ÿค– Generated with Claude Code (https://claude.com/claude-code) Co-Authored-By: Claude --- pyproject.toml | 2 +- src/pyaps/automation/WORKFLOW.md | 901 +++++++++++++++++++++++ src/pyaps/automation/__init__.py | 3 + src/pyaps/automation/types.py | 14 +- src/pyaps/automation/workflow.py | 447 +++++++++++ src/pyaps/automation/workflow_example.py | 595 +++++++++++++++ 6 files changed, 1959 insertions(+), 3 deletions(-) create mode 100644 src/pyaps/automation/WORKFLOW.md create mode 100644 src/pyaps/automation/workflow.py create mode 100644 src/pyaps/automation/workflow_example.py 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()