From 0f56ff975b02686960e07616868db97f95e9c35c Mon Sep 17 00:00:00 2001 From: hjnoh Date: Wed, 22 Oct 2025 20:44:03 +0900 Subject: [PATCH 1/2] FEAT: Add Data Management API examples Add comprehensive smoke test examples for Data Management API with auth integration: Hubs, Projects, Folders, Items, Versions, Buckets, Objects, and Commands. Also update __init__.py for easier imports. --- src/pyaps/datamanagement/__init__.py | 6 + src/pyaps/datamanagement/client.py | 314 +++++++++++++++++++++ src/pyaps/datamanagement/example.py | 391 +++++++++++++++++++++++++++ 3 files changed, 711 insertions(+) create mode 100644 src/pyaps/datamanagement/client.py create mode 100644 src/pyaps/datamanagement/example.py diff --git a/src/pyaps/datamanagement/__init__.py b/src/pyaps/datamanagement/__init__.py index e69de29..349b5b8 100644 --- a/src/pyaps/datamanagement/__init__.py +++ b/src/pyaps/datamanagement/__init__.py @@ -0,0 +1,6 @@ +# src/pyaps/datamanagement/__init__.py +from .client import DataManagementClient + +__all__ = [ + "DataManagementClient", +] diff --git a/src/pyaps/datamanagement/client.py b/src/pyaps/datamanagement/client.py new file mode 100644 index 0000000..ee40ef9 --- /dev/null +++ b/src/pyaps/datamanagement/client.py @@ -0,0 +1,314 @@ +# src/pyaps/datamanagement/client.py +from __future__ import annotations + +from typing import Any, Callable, Dict, Iterable, Optional +import requests + +from pyaps.http.client import HTTPClient, HTTPError + +DEFAULT_PROJECT_BASE = "https://developer.api.autodesk.com/project/v1" +DEFAULT_DATA_BASE = "https://developer.api.autodesk.com/data/v1" +DEFAULT_OSS_BASE = "https://developer.api.autodesk.com/oss/v2" + +class DataManagementClient: + """ + APS Data Management + - project/v1 : Hubs, Projects, TopFolders + - data/v1 : Folders, Items, Versions, Storage, Commands + - oss/v2 : Buckets, Objects (signed upload/download) + """ + def __init__( + self, + token_provider: Callable[[], str], + *, + project_base_url: str = DEFAULT_PROJECT_BASE, + data_base_url: str = DEFAULT_DATA_BASE, + oss_base_url: str = DEFAULT_OSS_BASE, + timeout: float = 30.0, + user_agent: str = "pyaps-dm", + session: Optional[requests.Session] = None, + ) -> None: + self.http_project = HTTPClient( + token_provider, base_url=project_base_url, + user_agent=user_agent, timeout=timeout, session=session + ) + self.http_data = HTTPClient( + token_provider, base_url=data_base_url, + user_agent=user_agent, timeout=timeout, session=session + ) + self.http_oss = HTTPClient( + token_provider, base_url=oss_base_url, + user_agent=user_agent, timeout=timeout, session=session + ) + + # public facades + self.hubs = _Hubs(self) + self.projects = _Projects(self) + self.folders = _Folders(self) + self.items = _Items(self) + self.versions = _Versions(self) + self.buckets = _Buckets(self) + self.objects = _Objects(self) + self.commands = _Commands(self) + +# ---------------------------- +# Project v1 — Hubs / Projects / TopFolders +# ---------------------------- +class _Hubs: + def __init__(self, cli: DataManagementClient): self.cli = cli + + def list(self, *, limit: int | None = None) -> Iterable[Dict]: + """GET /project/v1/hubs — list accessible hubs.""" + params = {"page[limit]": limit} if limit else None + page = self.cli.http_project.get("/hubs", params=params) + yield from self.cli.http_project.paginate(page) + + def get(self, hub_id: str) -> Dict: + """GET /project/v1/hubs/:hub_id""" + return self.cli.http_project.get(f"/hubs/{hub_id}") + + def list_projects(self, hub_id: str, *, limit: int | None = None) -> Iterable[Dict]: + """GET /project/v1/hubs/:hub_id/projects — projects in a hub.""" + params = {"page[limit]": limit} if limit else None + page = self.cli.http_project.get(f"/hubs/{hub_id}/projects", params=params) + yield from self.cli.http_project.paginate(page) + +class _Projects: + def __init__(self, cli: DataManagementClient): self.cli = cli + + def get(self, hub_id: str, project_id: str) -> Dict: + """GET /project/v1/hubs/:hub_id/projects/:project_id""" + return self.cli.http_project.get(f"/hubs/{hub_id}/projects/{project_id}") + + def top_folders(self, hub_id: str, project_id: str) -> Dict: + """GET /project/v1/hubs/:hub_id/projects/:project_id/topFolders""" + return self.cli.http_project.get(f"/hubs/{hub_id}/projects/{project_id}/topFolders") + +# ---------------------------- +# Data v1 — Folders / Items / Versions / Storage / Commands +# ---------------------------- +class _Folders: + def __init__(self, cli: DataManagementClient): self.cli = cli + + def get(self, project_id: str, folder_id: str) -> Dict: + """GET /data/v1/projects/:project_id/folders/:folder_id""" + return self.cli.http_data.get(f"/projects/{project_id}/folders/{folder_id}") + + def contents(self, project_id: str, folder_id: str, *, limit: int | None = None, include: str | None = None) -> Iterable[Dict]: + """GET /data/v1/projects/:project_id/folders/:folder_id/contents""" + params: Dict[str, Any] = {} + if limit: params["page[limit]"] = limit + if include: params["include"] = include + page = self.cli.http_data.get(f"/projects/{project_id}/folders/{folder_id}/contents", params=params) + yield from self.cli.http_data.paginate(page) + + def search(self, project_id: str, folder_id: str, q: str, *, limit: int | None = None) -> Iterable[Dict]: + """GET /data/v1/projects/:project_id/folders/:folder_id/search?q=...""" + params: Dict[str, Any] = {"q": q} + if limit: params["page[limit]"] = limit + page = self.cli.http_data.get(f"/projects/{project_id}/folders/{folder_id}/search", params=params) + yield from self.cli.http_data.paginate(page) + + def create(self, project_id: str, parent_folder_id: str, name: str, *, hidden:bool = False) -> Dict: + """POST /data/v1/projects/:project_id/folders""" + body = { + "data": { + "type": "folders", + "attributes": {"name": name, "hidden": hidden}, + "relationships": {"parent": {"data": {"type": "folders", "id": parent_folder_id}}}, + } + } + resp = self.cli.http_data.post(f"/projects/{project_id}/folders", json=body) + return (resp or {}).get("data", {}) + + def patch(self, project_id: str, folder_id: str, attributes: Dict) -> Dict: + """PATCH /data/v1/projects/:project_id/folders/:folder_id""" + body = {"data": {"type": "folders", "id": folder_id, "attributes": attributes}} + return self.cli.http_data.patch(f"/projects/{project_id}/folders/{folder_id}", json=body) + +class _Items: + def __init__(self, cli: DataManagementClient): self.cli = cli + + def get(self, project_id: str, item_id: str) -> Dict: + """GET /data/v1/projects/:project_id/items/:item_id""" + return self.cli.http_data.get(f"/projects/{project_id}/items/{item_id}") + + def list_versions(self, project_id: str, item_id: str, *, limit: int | None = None) -> Iterable[Dict]: + """GET /data/v1/projects/:project_id/items/:item_id/versions""" + params = {"page[limit]": limit} if limit else None + page = self.cli.http_data.get(f"/projects/{project_id}/items/{item_id}/versions", params=params) + yield from self.cli.http_data.paginate(page) + + def create_with_first_version(self, project_id: str, parent_folder_id: str, file_name: str, storage_urn: str) -> Dict: + """POST /data/v1/projects/:project_id/items — create item + first version""" + body = { + "data": { + "type": "items", + "attributes": {"displayName": file_name}, + "relationships": { + "tip": {"data": {"type": "versions", "id": "1"}}, + "parent": {"data": {"type": "folders", "id": parent_folder_id}}, + }, + }, + "included": [ + {"type": "versions", "id": "1", "attributes": {"name": file_name, "storageUrn": storage_urn}} + ], + } + resp = self.cli.http_data.post(f"/projects/{project_id}/items", json=body) + return (resp or {}).get("data", {}) + +class _Versions: + def __init__(self, cli: DataManagementClient): self.cli = cli + + def get(self, project_id: str, version_id: str) -> Dict: + """GET /data/v1/projects/:project_id/versions/:version_id""" + return self.cli.http_data.get(f"/projects/{project_id}/versions/{version_id}") + + def create(self, project_id: str, item_id: str, file_name: str, storage_urn: str) -> Dict: + """POST /data/v1/projects/:project_id/versions — create a new version for an item.""" + body = { + "data": { + "type": "versions", + "attributes": {"name": file_name, "storageUrn": storage_urn}, + "relationships": {"item": {"data": {"type": "items", "id": item_id}}}, + } + } + resp = self.cli.http_data.post(f"/projects/{project_id}/versions", json=body) + return (resp or {}).get("data", {}) + +class _Commands: + def __init__(self, cli: DataManagementClient): self.cli = cli + + def post(self, project_id: str, payload: Dict) -> Dict: + """POST /data/v1/projects/:project_id/commands""" + return self.cli.http_data.post(f"/projects/{project_id}/commands", json=payload) + +# ---------------------------- +# OSS v2 — Objects +# ---------------------------- + +class _Objects: + def __init__(self, cli: DataManagementClient): self.cli = cli + + # ----- Storage (Data v1 → direct S3 PUT) ----- + def create_storage(self, project_id: str, target_folder_id: str, file_name: str) -> Dict: + """POST /data/v1/projects/:project_id/storage — reserve upload target and get signed URL metadata.""" + body = { + "data": { + "type": "objects", + "attributes": {"name": file_name}, + "relationships": {"target": {"data": {"type": "folders", "id": target_folder_id}}}, + } + } + return self.cli.http_data.post(f"/projects/{project_id}/storage", json=body) + + def upload_via_storage(self, storage_resp: Dict, payload: bytes | str | Any, *, timeout: float | None = None) -> None: + """ + Use storage response's signed URL to upload in a single PUT. + """ + data = (storage_resp or {}).get("data", {}) or {} + attrs = data.get("attributes", {}) or {} + up = attrs.get("uploadParameters", {}) or {} + url = up.get("url") or ((data.get("links") or {}).get("signedUrl") or {}).get("href") + print("[datamanagement/client.py] url: ", url) + headers = up.get("headers") or {} + if not url: + raise HTTPError(500, "PUT", "signed-url", "No signed URL in storage response") + self.cli.http_data.put_signed_url(url, payload, headers=headers, timeout=timeout) + + # ----- OSS v2: object metadata/details ----- + def get_details(self, bucket_key: str, object_key: str) -> Dict: + """GET /oss/v2/buckets/:bucketKey/objects/:objectKey/details""" + return self.cli.http_oss.get(f"/buckets/{bucket_key}/objects/{object_key}/details") + + # ----- OSS v2: delete object ----- + def delete(self, bucket_key: str, object_key: str) -> None: + """DELETE /oss/v2/buckets/:bucketKey/objects/:objectKey""" + self.cli.http_oss.delete(f"/buckets/{bucket_key}/objects/{object_key}") + + # ----- OSS v2: copy object (within same bucket) ----- + def copy_to(self, bucket_key: str, object_key: str, new_object_key: str) -> Dict: + """PUT /oss/v2/buckets/:bucketKey/objects/:objectKey/copyto/:newObjectKey""" + return self.cli.http_oss.request_json( + "PUT", + f"/buckets/{bucket_key}/objects/{object_key}/copyto/{new_object_key}" + ) + + # ----- OSS v2: signed S3 upload/download ----- + def get_signed_upload(self, bucket_key: str, object_key: str, *, parts: int | None = None, useAcceleration: bool | None = None) -> Dict: + """GET /oss/v2/buckets/:bucketKey/objects/:objectKey/signeds3upload""" + params: Dict[str, Any] = {} + if parts is not None: params["parts"] = parts + if useAcceleration: params["useAcceleration"] = "true" + return self.cli.http_oss.get(f"/buckets/{bucket_key}/objects/{object_key}/signeds3upload", params=params) + + def complete_signed_upload(self, bucket_key: str, object_key: str, upload_key: str, *, size: int | None = None, etags: list[str] | None = None) -> Dict: + """POST /oss/v2/buckets/:bucketKey/objects/:objectKey/signeds3upload""" + body: Dict[str, Any] = {"uploadKey": upload_key} + if size is not None: body["size"] = size + if etags: body["eTags"] = etags + return self.cli.http_oss.post(f"/buckets/{bucket_key}/objects/{object_key}/signeds3upload", json=body) + + def get_signed_download(self, bucket_key: str, object_key: str, *, minutes_valid: int | None = None) -> Dict: + """GET /oss/v2/buckets/:bucketKey/objects/:objectKey/signeds3download""" + params = {"minutesExpiration": minutes_valid} if minutes_valid else None + return self.cli.http_oss.get(f"/buckets/{bucket_key}/objects/{object_key}/signeds3download", params=params) + + # ----- OSS v2: signed (proxy) upload/download ticket ----- + def post_signed(self, bucket_key: str, object_key: str, *, access: str = "readwrite", use_cookies: bool | None = None) -> Dict: + """ + POST /oss/v2/buckets/:bucketKey/objects/:objectKey/signed + - access: 'read', 'write', 'readwrite' + - 반환 값은 signedUrl(단건) 혹은 signedUrls(다건) 등을 포함 (문서에 따라) + """ + params: Dict[str, Any] = {"access": access} + if use_cookies is True: + params["useCookies"] = "true" + return self.cli.http_oss.post(f"/buckets/{bucket_key}/objects/{object_key}/signed", params=params, json={}) + + def upload_via_signed(self, signed_resp: Dict, file_path: str | bytes, *, timeout: float | None = None) -> None: + """ + POST /signed 응답의 signedUrl(또는 signedUrls[0]) 로 실제 바이트 업로드 (HTTP PUT) + """ + data = signed_resp or {} + url = (data.get("signedUrl") + or (isinstance(data.get("signedUrls"), list) and data["signedUrls"][0]) + or None) + if not url: + raise HTTPError(500, "PUT", "signed-url", "No signedUrl in POST /signed response") + self.cli.http_oss.put_signed_url(url, file_path, headers={}, timeout=timeout) + +# ---------------------------- +# OSS v2 — Buckets +# ---------------------------- +class _Buckets: + def __init__(self, cli: DataManagementClient): self.cli = cli + + def list(self, *, region: str | None = None, limit: int | None = None) -> Iterable[Dict]: + """GET /oss/v2/buckets — list buckets owned by the app.""" + params: Dict[str, Any] = {} + if region: params["region"] = region + if limit: params["page[limit]"] = limit + page = self.cli.http_oss.get("/buckets", params=params) + data = (page or {}).get("items") or (page or {}).get("data") or [] + for it in data: + yield it + + def create(self, bucket_key: str, *, region: str | None = None, policy_key: str | None = None) -> Dict: + """POST /oss/v2/buckets — create a bucket.""" + body: Dict[str, Any] = {"bucketKey": bucket_key} + if region: body["region"] = region + if policy_key: body["policyKey"] = policy_key + return self.cli.http_oss.post("/buckets", json=body) + + def get(self, bucket_key: str) -> Dict: + """GET /oss/v2/buckets/:bucketKey/details""" + return self.cli.http_oss.get(f"/buckets/{bucket_key}/details") + + def list_objects(self, bucket_key: str, *, limit: int | None = None) -> Iterable[Dict]: + """GET /oss/v2/buckets/:bucketKey/objects — list objects in a bucket.""" + params = {"page[limit]": limit} if limit else None + page = self.cli.http_oss.get(f"/buckets/{bucket_key}/objects", params=params) + data = (page or {}).get("items") or (page or {}).get("data") or [] + for it in data: + yield it \ No newline at end of file diff --git a/src/pyaps/datamanagement/example.py b/src/pyaps/datamanagement/example.py new file mode 100644 index 0000000..2d2f17b --- /dev/null +++ b/src/pyaps/datamanagement/example.py @@ -0,0 +1,391 @@ +""" +APS Data Management API 사용 예제 +Hubs, Projects, Folders, Items, Versions, Buckets, Objects 등의 기능을 smoke test 형태로 시연 +""" +from __future__ import annotations + +import os +from pathlib import Path +from pyaps.auth import AuthClient, Scopes, InMemoryTokenStore +from pyaps.datamanagement.client 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() + + +def create_dm_client() -> DataManagementClient: + """Create DataManagementClient with 3-legged token provider""" + auth_client = AuthClient( + client_id=os.getenv("APS_CLIENT_ID"), + client_secret=os.getenv("APS_CLIENT_SECRET"), + store=InMemoryTokenStore(), + ) + + # For Data Management, we typically use 2-legged with data:read/write scopes + def token_provider() -> str: + scopes = [Scopes.DATA_READ, Scopes.DATA_WRITE, Scopes.BUCKET_READ] + token = auth_client.two_legged.get_token(scopes) + return token.access_token + + return DataManagementClient(token_provider=token_provider) + + +def example_hubs(): + """Hubs API 예제""" + print("\n" + "="*60) + print("1. Hubs API") + print("="*60) + + dm = create_dm_client() + + # 1-1. List hubs + print("\n[1-1] List accessible hubs") + try: + hubs = list(dm.hubs.list(limit=10)) + print(f" ✓ Found {len(hubs)} hubs") + + if not hubs: + print(" ℹ No hubs found. You may need 3-legged authentication for BIM 360/ACC.") + return None, None + + for hub in hubs[:3]: + hub_data = hub.get('attributes', {}) + print(f" - {hub_data.get('name', 'N/A')} (ID: {hub.get('id', 'N/A')})") + + # 1-2. Get hub details + print("\n[1-2] Get hub details") + first_hub_id = hubs[0]['id'] + hub_detail = dm.hubs.get(first_hub_id) + hub_attrs = hub_detail.get('data', {}).get('attributes', {}) + print(f" ✓ Hub: {hub_attrs.get('name')}") + print(f" ✓ Region: {hub_attrs.get('region')}") + + # 1-3. List projects in hub + print("\n[1-3] List projects in hub") + projects = list(dm.hubs.list_projects(first_hub_id, limit=5)) + print(f" ✓ Found {len(projects)} projects") + for proj in projects[:3]: + proj_attrs = proj.get('attributes', {}) + print(f" - {proj_attrs.get('name', 'N/A')} (ID: {proj.get('id', 'N/A')})") + + return first_hub_id, projects[0]['id'] if projects else None + except Exception as e: + print(f" ✗ Error: {e}") + return None, None + + +def example_projects(hub_id: str, project_id: str): + """Projects API 예제""" + print("\n" + "="*60) + print("2. Projects API") + print("="*60) + + if not hub_id or not project_id: + print(" ℹ Skipped - no hub/project available") + return None + + dm = create_dm_client() + + # 2-1. Get project details + print("\n[2-1] Get project details") + try: + project = dm.projects.get(hub_id, project_id) + proj_data = project.get('data', {}) + proj_attrs = proj_data.get('attributes', {}) + print(f" ✓ Project: {proj_attrs.get('name')}") + print(f" ✓ Type: {proj_data.get('type')}") + + # 2-2. Get top folders + print("\n[2-2] Get top folders") + top_folders_resp = dm.projects.top_folders(hub_id, project_id) + top_folders = top_folders_resp.get('data', []) + print(f" ✓ Found {len(top_folders)} top folders") + for folder in top_folders[:3]: + folder_attrs = folder.get('attributes', {}) + print(f" - {folder_attrs.get('name', 'N/A')} (ID: {folder.get('id', 'N/A')})") + + return top_folders[0]['id'] if top_folders else None + except Exception as e: + print(f" ✗ Error: {e}") + return None + + +def example_folders(project_id: str, folder_id: str): + """Folders API 예제""" + print("\n" + "="*60) + print("3. Folders API") + print("="*60) + + if not project_id or not folder_id: + print(" ℹ Skipped - no project/folder available") + return + + dm = create_dm_client() + + # 3-1. Get folder details + print("\n[3-1] Get folder details") + try: + folder = dm.folders.get(project_id, folder_id) + folder_data = folder.get('data', {}) + folder_attrs = folder_data.get('attributes', {}) + print(f" ✓ Folder: {folder_attrs.get('name')}") + print(f" ✓ Hidden: {folder_attrs.get('hidden', False)}") + + # 3-2. List folder contents + print("\n[3-2] List folder contents") + contents = list(dm.folders.contents(project_id, folder_id, limit=10)) + print(f" ✓ Found {len(contents)} items") + for item in contents[:5]: + item_attrs = item.get('attributes', {}) + item_type = item.get('type', 'unknown') + print(f" - [{item_type}] {item_attrs.get('displayName') or item_attrs.get('name', 'N/A')}") + + # 3-3. Search in folder (structure only) + print("\n[3-3] Search in folder (structure only)") + print(" Usage:") + print(f" results = dm.folders.search('{project_id}', '{folder_id}', q='*.rvt')") + print(" for item in results:") + print(" print(item['attributes']['displayName'])") + + # 3-4. Create subfolder (structure only) + print("\n[3-4] Create subfolder (structure only)") + print(" Usage:") + print(f" new_folder = dm.folders.create(") + print(f" project_id='{project_id}',") + print(f" parent_folder_id='{folder_id}',") + print(f" name='New Subfolder'") + print(f" )") + + except Exception as e: + print(f" ✗ Error: {e}") + + +def example_items(): + """Items & Versions API 예제""" + print("\n" + "="*60) + print("4. Items & Versions API") + print("="*60) + + print("\n[4-1] Get item details (structure only)") + print(" Usage:") + print(" item = dm.items.get(project_id, item_id)") + print(" print(item['data']['attributes']['displayName'])") + + print("\n[4-2] List item versions (structure only)") + print(" Usage:") + print(" versions = dm.items.list_versions(project_id, item_id)") + print(" for version in versions:") + print(" print(version['attributes']['versionNumber'])") + + print("\n[4-3] Create item with first version (structure only)") + print(" Workflow:") + print(" 1. Create storage: storage = dm.objects.create_storage(project_id, folder_id, 'file.rvt')") + print(" 2. Upload file: dm.objects.upload_via_storage(storage, file_bytes)") + print(" 3. Create item: item = dm.items.create_with_first_version(") + print(" project_id, folder_id, 'file.rvt', storage['data']['id'])") + + print("\n[4-4] Create new version (structure only)") + print(" Usage:") + print(" new_version = dm.versions.create(project_id, item_id, 'file_v2.rvt', storage_urn)") + + +def example_buckets(): + """Buckets API 예제""" + print("\n" + "="*60) + print("5. Buckets API (OSS v2)") + print("="*60) + + dm = create_dm_client() + + # 5-1. List buckets + print("\n[5-1] List buckets") + try: + buckets = list(dm.buckets.list(limit=10)) + print(f" ✓ Found {len(buckets)} buckets") + for bucket in buckets[:3]: + bucket_key = bucket.get('bucketKey', 'N/A') + print(f" - {bucket_key}") + + # 5-2. Create bucket (structure only) + print("\n[5-2] Create bucket (structure only)") + print(" Usage:") + print(" new_bucket = dm.buckets.create(") + print(" bucket_key='my-unique-bucket-key',") + print(" region='US',") + print(" policy_key='transient' # or 'temporary', 'persistent'") + print(" )") + + if buckets: + # 5-3. Get bucket details + print("\n[5-3] Get bucket details") + first_bucket_key = buckets[0].get('bucketKey') + bucket = dm.buckets.get(first_bucket_key) + print(f" ✓ Bucket: {bucket.get('bucketKey')}") + print(f" ✓ Policy: {bucket.get('policyKey')}") + + # 5-4. List objects in bucket + print("\n[5-4] List objects in bucket") + objects = list(dm.buckets.list_objects(first_bucket_key, limit=5)) + print(f" ✓ Found {len(objects)} objects") + for obj in objects[:3]: + print(f" - {obj.get('objectKey', 'N/A')} ({obj.get('size', 0)} bytes)") + + except Exception as e: + print(f" ✗ Error: {e}") + + +def example_objects(): + """Objects API 예제""" + print("\n" + "="*60) + print("6. Objects API (Storage & Upload)") + print("="*60) + + print("\n[6-1] Create storage for upload (structure only)") + print(" Usage:") + print(" storage = dm.objects.create_storage(") + print(" project_id='b.project_id',") + print(" target_folder_id='urn:adsk.wipprod:fs.folder:xxxxx',") + print(" file_name='sample.rvt'") + print(" )") + + print("\n[6-2] Upload file via storage (structure only)") + print(" Usage:") + print(" with open('local_file.rvt', 'rb') as f:") + print(" dm.objects.upload_via_storage(storage, f.read())") + + print("\n[6-3] Get signed upload URL (OSS) (structure only)") + print(" Usage:") + print(" signed = dm.objects.get_signed_upload(") + print(" bucket_key='my-bucket',") + print(" object_key='path/to/file.txt',") + print(" parts=1 # single-part upload") + print(" )") + + print("\n[6-4] Complete signed upload (structure only)") + print(" Usage:") + print(" result = dm.objects.complete_signed_upload(") + print(" bucket_key='my-bucket',") + print(" object_key='path/to/file.txt',") + print(" upload_key=signed['uploadKey']") + print(" )") + + print("\n[6-5] Get signed download URL (structure only)") + print(" Usage:") + print(" download = dm.objects.get_signed_download(") + print(" bucket_key='my-bucket',") + print(" object_key='path/to/file.txt',") + print(" minutes_valid=60") + print(" )") + print(" # Then use download['url'] to download the file") + + +def example_commands(): + """Commands API 예제""" + print("\n" + "="*60) + print("7. Commands API (Advanced)") + print("="*60) + + print("\n[7-1] Move/Copy items (structure only)") + print(" Usage:") + print(" command = dm.commands.post(project_id, {") + print(" 'data': {") + print(" 'type': 'commands',") + print(" 'attributes': {") + print(" 'extension': {") + print(" 'type': 'commands:autodesk.core:MoveTo',") + print(" 'version': '1.0.0'") + print(" }") + print(" },") + print(" 'relationships': {") + print(" 'resources': {'data': [{'type': 'items', 'id': 'item_id'}]},") + print(" 'target': {'data': {'type': 'folders', 'id': 'target_folder_id'}}") + print(" }") + print(" }") + print(" })") + + +def example_workflow(): + """Complete workflow: Upload file to project""" + print("\n" + "="*60) + print("8. Complete Upload Workflow Example") + print("="*60) + + print("\nWorkflow to upload a file to APS project:") + print("\n Step 1: Get project and folder IDs") + print(" hubs = list(dm.hubs.list())") + print(" projects = list(dm.hubs.list_projects(hub_id))") + print(" top_folders = dm.projects.top_folders(hub_id, project_id)") + print(" folder_id = top_folders['data'][0]['id']") + + print("\n Step 2: Create storage location") + print(" storage = dm.objects.create_storage(") + print(" project_id=project_id,") + print(" target_folder_id=folder_id,") + print(" file_name='myfile.rvt'") + print(" )") + + print("\n Step 3: Upload file to storage") + print(" with open('myfile.rvt', 'rb') as f:") + print(" dm.objects.upload_via_storage(storage, f.read())") + + print("\n Step 4: Create item with first version") + print(" item = dm.items.create_with_first_version(") + print(" project_id=project_id,") + print(" parent_folder_id=folder_id,") + print(" file_name='myfile.rvt',") + print(" storage_urn=storage['data']['id']") + print(" )") + print(" print(f'✓ File uploaded: {item[\"id\"]}')") + + +def main(): + """모든 예제 실행""" + print("\n" + "="*60) + print("APS Data Management 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'") + + # 환경 변수 체크 + has_credentials = bool(os.getenv("APS_CLIENT_ID") and os.getenv("APS_CLIENT_SECRET")) + + if has_credentials: + print("\n✓ Credentials found - running live examples") + + # Live examples + hub_id, project_id = example_hubs() + if hub_id and project_id: + folder_id = example_projects(hub_id, project_id) + if folder_id: + example_folders(project_id, folder_id) + + example_buckets() + else: + print("\n⚠ Credentials not found - showing structure only") + + # Structure-only examples (always show) + example_items() + example_objects() + example_commands() + example_workflow() + + print("\n" + "="*60) + print("✓ All examples completed") + print("="*60 + "\n") + + +if __name__ == "__main__": + main() From a997236db4508d6e30437b027db53c8e4e864dd6 Mon Sep 17 00:00:00 2001 From: hjnoh Date: Wed, 22 Oct 2025 20:47:24 +0900 Subject: [PATCH 2/2] RELEASE: Bump version to 0.0.3 Add Data Management API support with comprehensive examples and update README. --- README.md | 26 +++++++++++++++++--------- pyproject.toml | 2 +- 2 files changed, 18 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index fd49800..6ea40c4 100644 --- a/README.md +++ b/README.md @@ -20,31 +20,39 @@ py-aps is a Python SDK that provides a simple and intuitive interface for intera ## Quick Start +### Authentication ```python from pyaps.auth import AuthClient, Scopes # 2-legged OAuth client = AuthClient(client_id="...", client_secret="...") token = client.two_legged.get_token([Scopes.DATA_READ]) +``` + +### Data Management +```python +from pyaps.datamanagement import DataManagementClient + +dm = DataManagementClient(token_provider=lambda: token.access_token) + +# List hubs and projects +hubs = list(dm.hubs.list()) +projects = list(dm.hubs.list_projects(hub_id)) -# 3-legged OAuth with PKCE -verifier, challenge = client.three_legged.generate_pkce_pair() -auth_url = client.three_legged.build_authorize_url( - scopes=[Scopes.DATA_READ, Scopes.USER_PROFILE_READ], - code_challenge=challenge -) -# After user authorizes: token = client.three_legged.exchange_code(code, code_verifier=verifier) +# Browse folders +contents = list(dm.folders.contents(project_id, folder_id)) ``` -For more examples, see `src/pyaps/auth/example.py`. +For more examples, see `src/pyaps/auth/example.py` and `src/pyaps/datamanagement/example.py`. ## Project Status -**Current version: v0.0.2** - Authentication API support added +**Current version: v0.0.3** - Data Management API support added This package is currently in early development. Active development is underway by **voidbox**. ### Version History +- **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) diff --git a/pyproject.toml b/pyproject.toml index 65ab88c..6b5452f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "py-aps" -version = "0.0.2" +version = "0.0.3" description = "Autodesk Platform Service APIs Python SDK" readme = "README.md" requires-python = ">=3.9"