diff --git a/README.md b/README.md index 6ea40c4..0d5a5e4 100644 --- a/README.md +++ b/README.md @@ -43,19 +43,40 @@ projects = list(dm.hubs.list_projects(hub_id)) contents = list(dm.folders.contents(project_id, folder_id)) ``` -For more examples, see `src/pyaps/auth/example.py` and `src/pyaps/datamanagement/example.py`. +### Design Automation +```python +from pyaps.automation import AutomationClient + +auto = AutomationClient(token_provider=lambda: token.access_token) + +# List engines +engines = auto.list_engines() + +# Create and execute workitem +workitem = auto.start_workitem({ + 'activityId': 'Owner.MyActivity+prod', + 'arguments': {...} +}) +``` + +For more examples, see `src/pyaps/auth/example.py`, `src/pyaps/datamanagement/example.py`, and `src/pyaps/automation/example.py`. ## Project Status -**Current version: v0.0.3** - Data Management API support added +**Current version: v0.0.4** - Design Automation API support added This package is currently in early development. Active development is underway by **voidbox**. -### Version History +
+Version History + +- **v0.0.4** - Added Design Automation API client (Engines, AppBundles, Activities, WorkItems) - **v0.0.3** - Added Data Management API client (Hubs, Projects, Folders, Items, Versions, Buckets, Objects) - **v0.0.2** - Added OAuth 2.0 authentication client with 2-legged/3-legged flows, PKCE support, and token management - **v0.0.1** - Initial package release (placeholder) +
+ ## Contributing We welcome bug reports and feature requests through [GitHub Issues](https://github.com/voidbox-ai/pyaps/issues). diff --git a/pyproject.toml b/pyproject.toml index 6b5452f..384a760 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "py-aps" -version = "0.0.3" +version = "0.0.4" description = "Autodesk Platform Service APIs Python SDK" readme = "README.md" requires-python = ">=3.9" diff --git a/src/pyaps/automation/__init__.py b/src/pyaps/automation/__init__.py index e69de29..5908397 100644 --- a/src/pyaps/automation/__init__.py +++ b/src/pyaps/automation/__init__.py @@ -0,0 +1,13 @@ +# src/pyaps/automation/__init__.py +from .client import AutomationClient, AutomationError, DEFAULT_AUTOMATION_SCOPES +from .types import WorkItemArgument, WorkItemSpec, AppBundleSpec, ActivitySpec + +__all__ = [ + "AutomationClient", + "AutomationError", + "DEFAULT_AUTOMATION_SCOPES", + "WorkItemArgument", + "WorkItemSpec", + "AppBundleSpec", + "ActivitySpec", +] diff --git a/src/pyaps/automation/client.py b/src/pyaps/automation/client.py new file mode 100644 index 0000000..763b1c8 --- /dev/null +++ b/src/pyaps/automation/client.py @@ -0,0 +1,175 @@ +# src/pyaps/automation/client.py +from __future__ import annotations +from typing import Any, Callable, Dict, Literal, Optional +from pathlib import Path +import requests + +from pyaps.http.client import HTTPClient, HTTPError # ← 공용 모듈 사용 + +AutomationRegion = Literal["us-east", "eu-west"] + +DEFAULT_AUTOMATION_SCOPES = [ + "code:all", "data:read", "data:write", "data:create", "bucket:read", "bucket:create", "bucket:update", +] + +class AutomationError(RuntimeError): + def __init__(self, message: str, status: int, payload: Any | None = None): + super().__init__(f"[{status}] {message}") + self.status = status + self.payload = payload + +class AutomationClient: + def __init__( + self, + token_provider: Callable[[], str], + *, + region: AutomationRegion = "us-east", + user_agent: str = "pyaps-automation", + timeout: float = 30.0, + session: Optional[requests.Session] = None, + ) -> None: + base_url = f"https://developer.api.autodesk.com/da/{region}/v3" + self.http = HTTPClient( + token_provider, + base_url=base_url, + user_agent=user_agent, + timeout=timeout, + session=session, + ) + + # ------- ForgeApps ------- + def get_me(self) -> Dict[str, Any]: + return self.http.get("/forgeapps/me") + + # ------- Engines ------- + def list_engines(self, *, page: Optional[int] = None, page_size: Optional[int] = None) -> Dict[str, Any]: + params: Dict[str, Any] = {} + if page is not None: params["page"] = page + if page_size is not None: params["pageSize"] = page_size + return self.http.get("/engines", params=params or None) + + # ------- AppBundles ------- + def create_appbundle(self, payload: Dict[str, Any]) -> Dict[str, Any]: + return self.http.post("/appbundles", json=payload) + + def list_appbundles(self) -> Dict[str, Any]: + return self.http.get("/appbundles") + + def get_appbundle(self, appbundle_id: str) -> Dict[str, Any]: + return self.http.get(f"/appbundles/{appbundle_id}") + + def delete_appbundle(self, appbundle_id: str) -> None: + self.http.delete(f"/appbundles/{appbundle_id}") + + def create_appbundle_alias(self, appbundle_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: + return self.http.post(f"/appbundles/{appbundle_id}/aliases", json=payload) + + def list_appbundle_aliases(self, appbundle_id: str) -> Dict[str, Any]: + return self.http.get(f"/appbundles/{appbundle_id}/aliases") + + def get_appbundle_alias_detail(self, appbundle_id: str, alias: str) -> Dict[str, Any]: + return self.http.get(f"/appbundles/{appbundle_id}/aliases/{alias}") + + def set_appbundle_alias(self, appbundle_id: str, alias: str, *, version: int) -> Dict[str, Any]: + return self.http.patch(f"/appbundles/{appbundle_id}/aliases/{alias}", json={"version": version}) + + def delete_appbundle_alias(self, appbundle_id: str, alias: str): + self.http.delete(f"/appbundles/{appbundle_id}/aliases/{alias}") + + def create_appbundle_version(self, appbundle_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: + return self.http.post(f"/appbundles/{appbundle_id}/versions", json=payload) + + def list_appbundle_versions(self, appbundle_id: str) -> Dict[str, Any]: + return self.http.get(f"/appbundles/{appbundle_id}/versions") + + def get_appbundle_version_detail(self, appbundle_id: str, version: str) -> Dict[str, Any]: + return self.http.get(f"/appbundles/{appbundle_id}/versions/{version}") + + def delete_appbundle_version(self, appbundle_id: str, version: str): + self.http.delete(f"/appbundles/{appbundle_id}/versions/{version}") + + # Presigned form upload (S3) + def upload_form_file(self, upload_parameters: Dict[str, Any], file_path: str | Path, *, timeout: Optional[float] = None) -> None: + endpoint = (upload_parameters or {}).get("endpointURL") + form_data = (upload_parameters or {}).get("formData") + if not endpoint or not form_data: + raise AutomationError("endpointURL/formData missing in uploadParameters", 400, upload_parameters) + try: + self.http.post_presigned_form(endpoint, form_data, str(file_path), timeout=timeout) + except HTTPError as e: + raise AutomationError("Upload failed", e.status, e.body) + + def upload_appbundle_zip_from_create(self, create_response: Dict[str, Any], zip_path: str | Path, *, timeout: Optional[float] = None) -> None: + self.upload_form_file(create_response.get("uploadParameters"), zip_path, timeout=timeout) + + def upload_appbundle_zip_from_version(self, version_response: Dict[str, Any], zip_path: str | Path, *, timeout: Optional[float] = None) -> None: + self.upload_form_file(version_response.get("uploadParameters"), zip_path, timeout=timeout) + + # ------- Activities ------- + def create_activity(self, payload: Dict[str, Any]) -> Dict[str, Any]: + return self.http.post("/activities", json=payload) + + def list_activities(self) -> Dict[str, Any]: + return self.http.get("/activities") + + def get_activity(self, activity_id: str) -> Dict[str, Any]: + return self.http.get(f"/activities/{activity_id}") + + def delete_activity(self, activity_id: str): + self.http.delete(f"/activities/{activity_id}") + + def create_activity_alias(self, activity_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: + return self.http.post(f"/activities/{activity_id}/aliases", json=payload) + + def list_activity_aliases(self, activity_id: str) -> Dict[str, Any]: + return self.http.get(f"/activities/{activity_id}/aliases") + + def get_activity_alias_detail(self, activity_id: str, alias: str) -> Dict[str, Any]: + return self.http.get(f"/activities/{activity_id}/aliases/{alias}") + + def set_activity_alias(self, activity_id: str, alias: str, *, version: int) -> Dict[str, Any]: + return self.http.patch(f"/activities/{activity_id}/aliases/{alias}", json={"version": version}) + + def delete_activity_alias(self, activity_id: str, alias: str): + self.http.delete(f"/activities/{activity_id}/aliases/{alias}") + + def create_activity_version(self, activity_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: + return self.http.post(f"/activities/{activity_id}/versions", json=payload) + + def list_activity_versions(self, activity_id: str) -> Dict[str, Any]: + return self.http.get(f"/activities/{activity_id}/versions") + + def get_activity_version(self, activity_id: str, version: int) -> Dict[str, Any]: + return self.http.get(f"/activities/{activity_id}/versions/{version}") + + def delete_activity_version(self, activity_id: str, version: int): + self.http.delete(f"/activities/{activity_id}/versions/{version}") + + # ------- WorkItems ------- + def start_workitem(self, spec: "WorkItemSpec | Dict[str, Any]") -> Dict[str, Any]: + body = spec.to_dict() if hasattr(spec, "to_dict") else spec + return self.http.post("/workitems", json=body) + + def get_workitem(self, workitem_id: str) -> Dict[str, Any]: + return self.http.get(f"/workitems/{workitem_id}") + + def cancel_workitem(self, workitem_id: str) -> None: + self.http.delete(f"/workitems/{workitem_id}") + + def create_workitems_batch(self, workitems: "list[dict] | list[WorkItemSpec]") -> dict: + def _to_dict(wi): return wi.to_dict() if hasattr(wi, "to_dict") else wi + payload = [_to_dict(w) for w in workitems] + return self.http.post("/workitems/batch", json=payload) + + def get_workitems_status(self, ids: "list[str]") -> dict: + return self.http.post("/workitems/status", json=ids) + + def combine_workitems(self, payload: dict) -> dict: + return self.http.post("/workitems/combine", json=payload) + + # ------- ServiceLimits ------- + def get_service_limits(self, owner: str) -> Dict[str, Any]: + return self.http.get(f"/servicelimits/{owner}") + + def put_service_limits(self, owner: str, payload: Dict[str, Any]) -> Dict[str, Any]: + return self.http.request_json("PUT", f"/servicelimits/{owner}", json=payload) \ No newline at end of file diff --git a/src/pyaps/automation/example.py b/src/pyaps/automation/example.py new file mode 100644 index 0000000..128a39e --- /dev/null +++ b/src/pyaps/automation/example.py @@ -0,0 +1,366 @@ +""" +APS Design Automation API 사용 예제 +AppBundles, Activities, WorkItems 등의 기능을 smoke test 형태로 시연 +""" +from __future__ import annotations + +import json +import os +import time +from pathlib import Path + +from pyaps.auth import AuthClient, Scopes, InMemoryTokenStore +from pyaps.automation.client import AutomationClient, DEFAULT_AUTOMATION_SCOPES +from pyaps.datamanagement import DataManagementClient +from pyaps.http.client import HTTPError + + +# 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 _print_hdr(title: str): + print(f"\n{title}") + print("-" * len(title)) + + +def _p(obj): + print(json.dumps(obj, indent=2, ensure_ascii=False)) + + +def create_clients(): + """Create auth and automation clients""" + 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-smoke", + timeout=30.0, + ) + + dm = DataManagementClient( + token_provider=token_provider, + user_agent="pyaps-automation-smoke", + timeout=30.0, + ) + + return auto, dm + + +def example_engines(): + """Engines API 예제""" + print("\n" + "="*60) + print("1. Engines API") + print("="*60) + + auto, _ = create_clients() + + # 1-1. List engines + print("\n[1-1] List available engines") + try: + engines = auto.list_engines(page=1, page_size=50) + data = engines.get('data', []) + print(f" ✓ Found {len(data)} engines") + for engine in data[:5]: + print(f" - {engine}") + except Exception as e: + print(f" ✗ Error: {e}") + + +def example_forgeapps(): + """ForgeApps API 예제""" + print("\n" + "="*60) + print("2. ForgeApps API") + print("="*60) + + auto, _ = create_clients() + + # 2-1. Get me + print("\n[2-1] Get current app information") + try: + me = auto.get_me() + _p(me) + except Exception as e: + print(f" ✗ Error: {e}") + + +def example_appbundles(): + """AppBundles API 예제""" + print("\n" + "="*60) + print("3. AppBundles API") + print("="*60) + + auto, _ = create_clients() + + # 3-1. List appbundles + print("\n[3-1] List appbundles") + try: + bundles = auto.list_appbundles() + data = bundles.get('data', []) + print(f" ✓ Found {len(data)} appbundles") + for bundle in data[:3]: + print(f" - {bundle}") + except Exception as e: + print(f" ✗ Error: {e}") + + # 3-2. Create appbundle (structure only) + print("\n[3-2] Create appbundle (structure only)") + print(" Usage:") + print(" appbundle = auto.create_appbundle({") + print(" 'id': 'MyAppBundle',") + print(" 'engine': 'Autodesk.Revit+2023',") + print(" 'description': 'My custom Revit addin'") + print(" })") + + # 3-3. Upload appbundle zip (structure only) + print("\n[3-3] Upload appbundle zip (structure only)") + print(" Workflow:") + print(" 1. Create appbundle: response = auto.create_appbundle({...})") + print(" 2. Upload zip: auto.upload_appbundle_zip_from_create(response, 'bundle.zip')") + print(" 3. Create alias: auto.create_appbundle_alias('MyAppBundle', {'id': 'prod', 'version': 1})") + + # 3-4. Manage versions (structure only) + print("\n[3-4] Manage appbundle versions (structure only)") + print(" Create new version:") + print(" version = auto.create_appbundle_version('MyAppBundle', {") + print(" 'engine': 'Autodesk.Revit+2023',") + print(" 'description': 'Version 2'") + print(" })") + print(" auto.upload_appbundle_zip_from_version(version, 'bundle_v2.zip')") + print(" auto.set_appbundle_alias('MyAppBundle', 'prod', version=2)") + + +def example_activities(): + """Activities API 예제""" + print("\n" + "="*60) + print("4. Activities API") + print("="*60) + + auto, _ = create_clients() + + # 4-1. List activities + print("\n[4-1] List activities") + try: + activities = auto.list_activities() + data = activities.get('data', []) + print(f" ✓ Found {len(data)} activities") + for activity in data[:3]: + print(f" - {activity}") + except Exception as e: + print(f" ✗ Error: {e}") + + # 4-2. Create activity (structure only) + print("\n[4-2] Create activity (structure only)") + print(" Usage:") + print(" activity = auto.create_activity({") + print(" 'id': 'MyActivity',") + print(" 'engine': 'Autodesk.Revit+2023',") + print(" 'commandLine': [") + print(" '$(engine.path)\\\\revitcoreconsole.exe /i \"$(args[inputFile].path)\" /al \"$(appbundles[MyBundle].path)\"'") + print(" ],") + print(" 'parameters': {") + print(" 'inputFile': {'verb': 'get', 'description': 'Input RVT file'},") + print(" 'outputFile': {'verb': 'put', 'description': 'Output RVT file'}") + print(" },") + print(" 'appbundles': ['Owner.MyAppBundle+prod']") + print(" })") + + # 4-3. Manage activity versions (structure only) + print("\n[4-3] Manage activity versions (structure only)") + print(" Create alias:") + print(" auto.create_activity_alias('MyActivity', {'id': 'prod', 'version': 1})") + print(" Create new version:") + print(" auto.create_activity_version('MyActivity', {...})") + print(" Update alias:") + print(" auto.set_activity_alias('MyActivity', 'prod', version=2)") + + +def example_workitems(): + """WorkItems API 예제""" + print("\n" + "="*60) + print("5. WorkItems API") + print("="*60) + + print("\n[5-1] Start workitem (structure only)") + print(" Workflow:") + print(" # 1. Prepare input/output signed URLs (using OSS)") + print(" input_url = '...' # signed download URL") + print(" output_url = '...' # signed upload URL") + print() + print(" # 2. Start workitem") + print(" workitem = auto.start_workitem({") + print(" 'activityId': 'Owner.MyActivity+prod',") + print(" 'arguments': {") + print(" 'inputFile': {'url': input_url, 'verb': 'get'},") + print(" 'outputFile': {'url': output_url, 'verb': 'put'}") + print(" }") + print(" })") + print(" workitem_id = workitem['id']") + print() + print(" # 3. Poll workitem status") + print(" while True:") + print(" status = auto.get_workitem(workitem_id)") + print(" if status['status'] in ('success', 'failed', 'cancelled'):") + print(" break") + print(" time.sleep(10)") + + print("\n[5-2] Batch workitems (structure only)") + print(" Usage:") + print(" workitems = [") + print(" {'activityId': '...', 'arguments': {...}},") + print(" {'activityId': '...', 'arguments': {...}}") + print(" ]") + print(" batch = auto.create_workitems_batch(workitems)") + print(" batch_ids = [wi['id'] for wi in batch]") + print(" status = auto.get_workitems_status(batch_ids)") + + print("\n[5-3] Cancel workitem (structure only)") + print(" Usage:") + print(" auto.cancel_workitem(workitem_id)") + + +def example_service_limits(): + """Service Limits API 예제""" + print("\n" + "="*60) + print("6. Service Limits API") + print("="*60) + + print("\n[6-1] Get service limits (structure only)") + print(" Usage:") + print(" limits = auto.get_service_limits('owner_id')") + print(" print(limits)") + print() + print(" Example response:") + print(" {") + print(" 'maxConcurrentWorkitems': 10,") + print(" 'maxWorkitemDuration': 3600") + print(" }") + + +def example_complete_workflow(): + """Complete workflow example""" + print("\n" + "="*60) + print("7. Complete Workflow Example") + print("="*60) + + print("\nComplete Design Automation workflow:") + print() + print("Step 1: Create and upload AppBundle") + print(" appbundle = auto.create_appbundle({") + print(" 'id': 'MyBundle',") + print(" 'engine': 'Autodesk.Revit+2023',") + print(" 'description': 'My Revit addin'") + print(" })") + print(" auto.upload_appbundle_zip_from_create(appbundle, 'bundle.zip')") + print(" auto.create_appbundle_alias('MyBundle', {'id': 'prod', 'version': 1})") + print() + print("Step 2: Create Activity") + print(" activity = auto.create_activity({") + print(" 'id': 'MyActivity',") + print(" 'engine': 'Autodesk.Revit+2023',") + print(" 'commandLine': ['...'],") + print(" 'parameters': {...},") + print(" 'appbundles': ['Owner.MyBundle+prod']") + print(" })") + print(" auto.create_activity_alias('MyActivity', {'id': 'prod', 'version': 1})") + print() + print("Step 3: Prepare input/output files (using Data Management)") + print(" # Upload input file to OSS") + print(" signed_input = dm.objects.get_signed_download(bucket, 'input.rvt')") + print(" input_url = signed_input['url']") + print() + print(" # Prepare output upload URL") + print(" signed_output = dm.objects.get_signed_upload(bucket, 'output.rvt')") + print(" output_url = signed_output['urls'][0]") + print() + print("Step 4: Execute WorkItem") + print(" workitem = auto.start_workitem({") + print(" 'activityId': 'Owner.MyActivity+prod',") + print(" 'arguments': {") + print(" 'inputFile': {'url': input_url, 'verb': 'get'},") + print(" 'outputFile': {'url': output_url, 'verb': 'put'}") + print(" }") + print(" })") + print() + print("Step 5: Monitor and download results") + print(" # Poll until completion") + print(" status = auto.get_workitem(workitem['id'])") + print(" # Download output file from OSS when status == 'success'") + + +def main(): + """모든 예제 실행""" + print("\n" + "="*60) + print("APS Design Automation API - Smoke Test Examples") + print("="*60) + print("\nℹ Set environment variables before running:") + 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'") + + # 환경 변수 체크 + has_credentials = bool(APS_CLIENT_ID and APS_CLIENT_SECRET) + + if has_credentials: + print("\n✓ Credentials found - running live examples") + + # Live examples + example_forgeapps() + example_engines() + example_appbundles() + example_activities() + else: + print("\n⚠ Credentials not found - showing structure only") + + # Structure-only examples (always show) + example_workitems() + example_service_limits() + example_complete_workflow() + + print("\n" + "="*60) + print("✓ All examples completed") + print("="*60) + print("\n📚 For complete AppBundle workflow, you'll need:") + print(" - AppBundle zip file (Revit/Inventor/AutoCAD plugin)") + print(" - Input files stored in OSS") + print(" - Activity definition matching your AppBundle") + print("\n") + + +if __name__ == "__main__": + try: + main() + except HTTPError as e: + print(f"\nHTTPError: [{e.status}] {e.method} {e.url}") + if e.body: + print(f"Response: {e.body[:500]}") + raise + except Exception as e: + print(f"\nError: {e}") + raise diff --git a/src/pyaps/automation/types.py b/src/pyaps/automation/types.py new file mode 100644 index 0000000..468c3c8 --- /dev/null +++ b/src/pyaps/automation/types.py @@ -0,0 +1,99 @@ +# src/pyaps/automation/types.py +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Literal, Optional + +@dataclass +class WorkItemArgument: + """ + Automation WorkItem 인자(입력/출력 공통 포맷) + """ + url: str + verb: Literal["get", "put", "head"] = "get" + headers: Optional[Dict[str, str]] = None + local_name: Optional[str] = None + on_demand: Optional[bool] = None + unzip: Optional[bool] = None + description: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + d: Dict[str, Any] = { + "url": self.url, + "verb": self.verb, + } + if self.headers is not None: + d["headers"] = self.headers + if self.local_name is not None: + d["localName"] = self.local_name + if self.on_demand is not None: + d["onDemand"] = self.on_demand + # unzip=True -> 서버에서 압축 해제 (zip=False로 표기) + if self.unzip is not None: + d["zip"] = not self.unzip + if self.description is not None: + d["description"] = self.description + return d + +@dataclass +class WorkItemSpec: + """ + WorkItem 생성 요청 + - activity_id 예시: '{nickname}.{activity}+{alias}' 또는 '{owner}.{activity}+{alias}' + - arguments: Activity에서 선언한 파라미터 이름을 key로 사용 + """ + activity_id: str + arguments: Dict[str, WorkItemArgument] = field(default_factory=dict) + nickname: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + return { + "activityId": self.activity_id, + "arguments": {k: v.to_dict() for k, v in self.arguments.items()}, + **({"nickname": self.nickname} if self.nickname else {}), + } + +@dataclass +class AppBundleSpec: + """ + AppBundle 생성/버전 생성 시 사용되는 사양의 간단 래퍼 + - 실제 API는 업로드 사전서명(Form) 방식 등을 반환할 수 있으므로, + 여기서는 최소 필드만 캡슐화하고, 나머지는 dict로 직접 전달하도록 설계해야 함 + """ + id: str + engine: str + description: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + d: Dict[str, Any] = { + "id": self.id, + "engine": self.engine, + } + if self.description: + d["description"] = self.description + return d + +@dataclass +class ActivitySpec: + """ + Activity 생성/버전 생성용 사양(유연성을 위해 dict 병행 권장) + """ + id: str + engine: str + command_line: List[str] + parameters: Dict[str, Any] = field(default_factory=dict) + appbundles: Optional[List[str]] = None + description: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + d: Dict[str, Any] = { + "id": self.id, + "engine": self.engine, + "commandLine": self.command_line, + "parameters": self.parameters, + } + if self.appbundles is not None: + d["appbundles"] = self.appbundles + if self.description: + d["description"] = self.description + return d \ No newline at end of file