Create Assets
This guide shows how to create assets (spreadsheets, documents, and folders) using the Python SDK. Assets are the core building blocks of the Athena Intelligence platform, allowing you to organize and manage your workspace programmatically.
Why Create Assets Programmatically? Automate workspace setup, generate dynamic reports, build custom workflows, and integrate Athena into your applications with proper asset organization.
Supported Asset Types:
spreadsheet- Athena spreadsheet with real-time collaborationdocument- Athena document for rich text editingfolder- Folder for organizing assets
Key features:
- Three core asset types - Create spreadsheets, documents, and folders
- Folder organization - Structure assets hierarchically in folders
- Batch operations - Create multiple assets efficiently
- Full type hints - Complete type safety with Python type hints
- Error handling - Comprehensive error handling for production use
Set Up Client
from athena import Athenafrom athena.core import ApiError# Production client setupclient = Athena(api_key="YOUR_API_KEY")# Custom API endpoint (if needed)custom_client = Athena(api_key="YOUR_API_KEY",base_url="https://your-custom-api.example.com")# Local developmentdev_client = Athena(api_key="YOUR_API_KEY",base_url="http://localhost:8000")
Type Definitions
Define proper types for better code organization:
from typing import TypedDict, Optional, List, Literalfrom dataclasses import dataclass# Supported asset typesAssetType = Literal["spreadsheet","document","folder"]# Asset creation result@dataclassclass AssetCreationResult:asset_id: strtitle: strasset_type: strcreated_at: strparent_folder_id: Optional[str] = Nonesuccess: bool = Trueerror: Optional[str] = None# Batch creation result@dataclassclass BatchAssetResult:successful: List[AssetCreationResult]failed: List[dict]total_created: inttotal_failed: int
Basic Asset Creation
Create a single asset with proper error handling:
def create_asset(asset_type: AssetType,title: Optional[str] = None,parent_folder_id: Optional[str] = None) -> AssetCreationResult:"""Create a single asset with error handling.Args:asset_type: Type of asset to create (spreadsheet, document, folder, etc.)title: Optional title for the assetparent_folder_id: Optional parent folder to create the asset inReturns:AssetCreationResult with success status and asset details"""try:print(f"📝 Creating {asset_type}...")response = client.assets.create(asset_type=asset_type,title=title,parent_folder_id=parent_folder_id)print("✅ Asset created successfully")print(f"Asset ID: {response.asset_id}")print(f"Title: {response.title}")print(f"Created at: {response.created_at}")return AssetCreationResult(asset_id=response.asset_id,title=response.title,asset_type=response.asset_type,created_at=response.created_at,parent_folder_id=response.parent_folder_id,success=True)except ApiError as e:error_message = f"API Error ({e.status_code}): {str(e)}"print(f"❌ Asset creation failed: {error_message}")raise Exception(error_message)except Exception as e:error_message = f"Asset creation failed: {str(e)}"print(f"❌ {error_message}")raise Exception(error_message)# Usage examplesdef basic_examples():"""Basic asset creation examples."""# Create a spreadsheetspreadsheet = create_asset("spreadsheet", "Q1 2024 Sales Report")# Create a documentdocument = create_asset("document", "Meeting Notes")# Create a folderfolder = create_asset("folder", "Project Alpha")return {"spreadsheet": spreadsheet,"document": document,"folder": folder}
Create Assets in Folders
Organize assets hierarchically with folders:
def create_asset_in_folder(asset_type: AssetType,title: str,parent_folder_id: str) -> AssetCreationResult:"""Create an asset inside a specific folder.Args:asset_type: Type of asset to createtitle: Title for the assetparent_folder_id: ID of the parent folderReturns:AssetCreationResult with creation details"""try:response = client.assets.create(asset_type=asset_type,title=title,parent_folder_id=parent_folder_id)print(f'✅ Created {asset_type} "{title}" in folder {parent_folder_id}')return AssetCreationResult(asset_id=response.asset_id,title=response.title,asset_type=response.asset_type,created_at=response.created_at,parent_folder_id=response.parent_folder_id,success=True)except Exception as e:error_message = str(e)return AssetCreationResult(asset_id="",title=title,asset_type=asset_type,created_at="",parent_folder_id=parent_folder_id,success=False,error=error_message)def create_project_structure(project_name: str) -> dict:"""Create an organized folder structure for a project.Args:project_name: Name of the projectReturns:Dictionary with project folder and subfolders"""print(f'📁 Creating project structure for "{project_name}"...')# Create main project folderproject_folder = create_asset("folder", project_name)if not project_folder.success:raise Exception(f"Failed to create project folder: {project_folder.error}")print(f"✅ Created project folder: {project_folder.asset_id}")# Create subfolderssubfolders_to_create = [{"name": "Documents", "type": "folder"},{"name": "Spreadsheets", "type": "folder"},{"name": "Reports", "type": "folder"},]subfolders = [create_asset_in_folder(subfolder["type"],subfolder["name"],project_folder.asset_id)for subfolder in subfolders_to_create]successful_subfolders = [f for f in subfolders if f.success]print(f"✅ Created {len(successful_subfolders)} subfolders")return {"project_folder": project_folder,"subfolders": subfolders}
Batch Asset Creation
Create multiple assets efficiently with error handling:
from typing import Listimport asynciofrom concurrent.futures import ThreadPoolExecutordef create_multiple_assets(requests: List[dict]) -> BatchAssetResult:"""Create multiple assets with parallel execution.Args:requests: List of dicts with asset_type, title, and optional parent_folder_idReturns:BatchAssetResult with successful and failed creations"""print(f"🔄 Creating {len(requests)} assets...")successful: List[AssetCreationResult] = []failed: List[dict] = []for req in requests:try:response = client.assets.create(asset_type=req["asset_type"],title=req.get("title"),parent_folder_id=req.get("parent_folder_id"))successful.append(AssetCreationResult(asset_id=response.asset_id,title=response.title,asset_type=response.asset_type,created_at=response.created_at,parent_folder_id=response.parent_folder_id,success=True))except Exception as e:failed.append({"request": req,"error": str(e)})print(f"✅ Successfully created: {len(successful)}")print(f"❌ Failed: {len(failed)}")return BatchAssetResult(successful=successful,failed=failed,total_created=len(successful),total_failed=len(failed))def create_quarterly_reports():"""Example: Create multiple quarterly reports."""quarters = ["Q1", "Q2", "Q3", "Q4"]year = "2024"requests = [{"asset_type": "spreadsheet","title": f"{quarter} {year} Sales Report"}for quarter in quarters]result = create_multiple_assets(requests)print("\n=== Batch Creation Results ===")print(f"Total created: {result.total_created}")print(f"Total failed: {result.total_failed}")if result.failed:print("\nFailed creations:")for item in result.failed:print(f" - {item['request']['title']}: {item['error']}")return result
Advanced Folder Organization
Create complex folder hierarchies:
from typing import Dict, Anyclass FolderStructure(TypedDict, total=False):name: strtype: AssetTypechildren: List['FolderStructure']class CreatedAssetNode(TypedDict, total=False):asset_id: strtitle: strasset_type: strchildren: List['CreatedAssetNode']def create_folder_hierarchy(structure: FolderStructure,parent_folder_id: Optional[str] = None) -> CreatedAssetNode:"""Recursively create a folder hierarchy from a structure definition.Args:structure: Folder structure definitionparent_folder_id: Optional parent folder IDReturns:CreatedAssetNode with created asset details"""# Create the current folder/assetresponse = client.assets.create(asset_type=structure["type"],title=structure["name"],parent_folder_id=parent_folder_id)print(f'✅ Created {structure["type"]}: {structure["name"]}')node: CreatedAssetNode = {"asset_id": response.asset_id,"title": response.title,"asset_type": response.asset_type}# Recursively create childrenif "children" in structure and structure["children"]:child_nodes = [create_folder_hierarchy(child, response.asset_id)for child in structure["children"]]node["children"] = child_nodesreturn nodedef create_complete_project_structure():"""Example: Create a complete project structure."""project_structure: FolderStructure = {"name": "Product Launch 2024","type": "folder","children": [{"name": "Research","type": "folder","children": [{"name": "Market Analysis", "type": "spreadsheet"},{"name": "Competitor Research", "type": "document"},{"name": "User Survey Results", "type": "spreadsheet"},]},{"name": "Planning","type": "folder","children": [{"name": "Timeline", "type": "spreadsheet"},{"name": "Budget", "type": "spreadsheet"},{"name": "Strategy Doc", "type": "document"},]},{"name": "Deliverables","type": "folder","children": [{"name": "Launch Plan", "type": "document"},{"name": "Marketing Materials", "type": "folder"},]}]}print("🏗️ Creating complete project structure...")result = create_folder_hierarchy(project_structure)print("🎉 Project structure created successfully!")return result
Error Handling and Retry Logic
Implement comprehensive error handling for production:
import timefrom typing import Optionaldef create_asset_with_retry(asset_type: AssetType,title: str,parent_folder_id: Optional[str] = None,max_retries: int = 3) -> AssetCreationResult:"""Create an asset with automatic retry logic.Args:asset_type: Type of asset to createtitle: Title for the assetparent_folder_id: Optional parent folder IDmax_retries: Maximum number of retry attemptsReturns:AssetCreationResult with creation details"""last_error: Optional[Exception] = Nonefor attempt in range(1, max_retries + 1):try:print(f"🔄 Attempt {attempt}/{max_retries} to create {asset_type}")response = client.assets.create(asset_type=asset_type,title=title,parent_folder_id=parent_folder_id)print(f"✅ Asset created successfully on attempt {attempt}")return AssetCreationResult(asset_id=response.asset_id,title=response.title,asset_type=response.asset_type,created_at=response.created_at,parent_folder_id=response.parent_folder_id,success=True)except ApiError as e:last_error = e# Don't retry on client errors (4xx)if 400 <= e.status_code < 500:print(f"❌ Client error ({e.status_code}): {str(e)}")return AssetCreationResult(asset_id="",title=title,asset_type=asset_type,created_at="",parent_folder_id=parent_folder_id,success=False,error=f"Client error ({e.status_code}): {str(e)}")# Retry on server errors or network issuesif attempt < max_retries:delay = min(1.0 * (2 ** (attempt - 1)), 10.0)print(f"⏳ Waiting {delay}s before retry...")time.sleep(delay)except Exception as e:last_error = eif attempt < max_retries:delay = min(1.0 * (2 ** (attempt - 1)), 10.0)print(f"⏳ Waiting {delay}s before retry...")time.sleep(delay)print("❌ Asset creation failed after all retry attempts")error_message = str(last_error) if last_error else "Unknown error"return AssetCreationResult(asset_id="",title=title,asset_type=asset_type,created_at="",parent_folder_id=parent_folder_id,success=False,error=f"Failed after {max_retries} attempts: {error_message}")
Type-Safe Asset Factory
Create a factory class for managing asset creation:
from typing import Dict, Anyclass AssetFactory:"""Factory class for creating different types of assets."""def __init__(self, api_key: str, base_url: Optional[str] = None):"""Initialize the asset factory.Args:api_key: Athena API keybase_url: Optional custom base URL"""self.client = Athena(api_key=api_key, base_url=base_url)def create_spreadsheet(self,title: str,parent_folder_id: Optional[str] = None) -> Dict[str, Any]:"""Create a spreadsheet asset."""return self._create_asset("spreadsheet", title, parent_folder_id)def create_document(self,title: str,parent_folder_id: Optional[str] = None) -> Dict[str, Any]:"""Create a document asset."""return self._create_asset("document", title, parent_folder_id)def create_folder(self,title: str,parent_folder_id: Optional[str] = None) -> Dict[str, Any]:"""Create a folder asset."""return self._create_asset("folder", title, parent_folder_id)def _create_asset(self,asset_type: AssetType,title: str,parent_folder_id: Optional[str] = None) -> Dict[str, Any]:"""Internal method to create an asset."""try:response = self.client.assets.create(asset_type=asset_type,title=title,parent_folder_id=parent_folder_id)print(f'✅ Created {asset_type}: {response.title} ({response.asset_id})')return {"asset_id": response.asset_id,"title": response.title,"asset_type": response.asset_type,"created_at": response.created_at,"parent_folder_id": response.parent_folder_id}except ApiError as e:raise Exception(f"Failed to create {asset_type}: {e.status_code} - {str(e)}")def create_batch(self,requests: List[Dict[str, Any]]) -> BatchAssetResult:"""Create multiple assets in batch.Args:requests: List of asset creation requestsReturns:BatchAssetResult with successful and failed creations"""print(f"🔄 Creating {len(requests)} assets in batch...")successful: List[AssetCreationResult] = []failed: List[dict] = []for req in requests:try:response = self._create_asset(req["type"],req["title"],req.get("parent_folder_id"))successful.append(AssetCreationResult(asset_id=response["asset_id"],title=response["title"],asset_type=response["asset_type"],created_at=response["created_at"],parent_folder_id=response.get("parent_folder_id"),success=True))except Exception as e:failed.append({"request": req,"error": str(e)})return BatchAssetResult(successful=successful,failed=failed,total_created=len(successful),total_failed=len(failed))# Usagedef factory_example():"""Example using the asset factory."""factory = AssetFactory(api_key="YOUR_API_KEY")# Create individual assetsspreadsheet = factory.create_spreadsheet("Sales Dashboard")document = factory.create_document("Product Requirements")folder = factory.create_folder("Marketing Campaign")# Create batch of assetsbatch_result = factory.create_batch([{"type": "spreadsheet", "title": "Budget 2024"},{"type": "document", "title": "Project Brief"},{"type": "folder", "title": "Assets"},])print(f"Created {batch_result.total_created} assets")return {"individual": [spreadsheet, document, folder],"batch": batch_result}
Complete Production Example
Here’s a production-ready asset management class:
import osfrom typing import Optional, List, Callablefrom dataclasses import dataclass, field@dataclassclass AssetCreationOptions:"""Options for asset creation."""retries: int = 3timeout: int = 60throw_on_error: bool = Falseparent_folder_id: Optional[str] = Noneclass AssetManager:"""Production-ready asset manager with comprehensive error handling."""def __init__(self, api_key: str, base_url: Optional[str] = None):"""Initialize the asset manager.Args:api_key: Athena API keybase_url: Optional custom base URL"""self.client = Athena(api_key=api_key, base_url=base_url)self.default_retries = 3def create(self,asset_type: AssetType,title: str,options: Optional[AssetCreationOptions] = None) -> AssetCreationResult:"""Create an asset with comprehensive error handling.Args:asset_type: Type of asset to createtitle: Title for the assetoptions: Creation options including retries and parent folderReturns:AssetCreationResult with creation details"""if options is None:options = AssetCreationOptions()retries = options.retries or self.default_retriesparent_folder_id = options.parent_folder_idthrow_on_error = options.throw_on_errorattempt = 0last_error: Optional[Exception] = Nonewhile attempt < retries:try:attempt += 1print(f"📝 Creating {asset_type} (attempt {attempt}/{retries})...")response = self.client.assets.create(asset_type=asset_type,title=title,parent_folder_id=parent_folder_id)print(f"✅ Asset created: {response.asset_id}")return AssetCreationResult(asset_id=response.asset_id,title=response.title,asset_type=response.asset_type,created_at=response.created_at,parent_folder_id=response.parent_folder_id,success=True)except ApiError as e:last_error = e# Don't retry on client errorsif 400 <= e.status_code < 500:print(f"❌ Client error ({e.status_code}): {str(e)}")if throw_on_error:raisereturn AssetCreationResult(asset_id="",title=title,asset_type=asset_type,created_at="",parent_folder_id=parent_folder_id,success=False,error=f"Client error ({e.status_code}): {str(e)}")print(f"⚠️ Attempt {attempt} failed: {str(e)}")# Wait before retry with exponential backoffif attempt < retries:delay = min(1.0 * (2 ** (attempt - 1)), 10.0)time.sleep(delay)except Exception as e:last_error = eprint(f"⚠️ Attempt {attempt} failed: {str(e)}")if attempt < retries:delay = min(1.0 * (2 ** (attempt - 1)), 10.0)time.sleep(delay)error_message = str(last_error) if last_error else "Unknown error"if throw_on_error:raise Exception(f"Asset creation failed after {retries} attempts: {error_message}")return AssetCreationResult(asset_id="",title=title,asset_type=asset_type,created_at="",parent_folder_id=parent_folder_id,success=False,error=f"Failed after {retries} attempts: {error_message}")def create_from_template(self,template: FolderStructure,parent_folder_id: Optional[str] = None) -> CreatedAssetNode:"""Create a workspace structure from a template.Args:template: Folder structure templateparent_folder_id: Optional parent folder IDReturns:CreatedAssetNode with created structure"""print(f'🏗️ Creating structure from template: {template["name"]}')response = self.create(template["type"],template["name"],AssetCreationOptions(parent_folder_id=parent_folder_id,throw_on_error=True))node: CreatedAssetNode = {"asset_id": response.asset_id,"title": response.title,"asset_type": response.asset_type}if "children" in template and template["children"]:print(f'📂 Creating {len(template["children"])} children...')child_nodes = [self.create_from_template(child, response.asset_id)for child in template["children"]]node["children"] = child_nodesreturn nodedef create_batch_with_progress(self,requests: List[Dict[str, Any]],on_progress: Optional[Callable[[int, int], None]] = None) -> BatchAssetResult:"""Create multiple assets with progress tracking.Args:requests: List of asset creation requestson_progress: Optional callback for progress updatesReturns:BatchAssetResult with successful and failed creations"""total = len(requests)print(f"🔄 Creating {total} assets with progress tracking...")successful: List[AssetCreationResult] = []failed: List[dict] = []for i, req in enumerate(requests):try:response = self.client.assets.create(asset_type=req["asset_type"],title=req.get("title"),parent_folder_id=req.get("parent_folder_id"))successful.append(AssetCreationResult(asset_id=response.asset_id,title=response.title,asset_type=response.asset_type,created_at=response.created_at,parent_folder_id=response.parent_folder_id,success=True))if on_progress:on_progress(len(successful), total)except Exception as e:failed.append({"request": req,"error": str(e)})return BatchAssetResult(successful=successful,failed=failed,total_created=len(successful),total_failed=len(failed))def production_example():"""Complete production usage example."""manager = AssetManager(api_key=os.getenv("ATHENA_API_KEY"))try:# Create a single asset with retrydocument = manager.create("document","Important Document",AssetCreationOptions(retries=5))if document.success:print(f"✅ Document created: {document.asset_id}")# Create from templateproject_template: FolderStructure = {"name": "Q1 2024 Initiative","type": "folder","children": [{"name": "Planning","type": "folder","children": [{"name": "Project Plan", "type": "document"},{"name": "Budget Tracker", "type": "spreadsheet"},]},{"name": "Execution","type": "folder","children": [{"name": "Tasks", "type": "spreadsheet"},{"name": "Progress Report", "type": "document"},]}]}project = manager.create_from_template(project_template)print(f"✅ Project structure created: {project['asset_id']}")# Batch creation with progressbatch_requests = [{"asset_type": "spreadsheet", "title": "Sales Data"},{"asset_type": "document", "title": "Analysis Report"},{"asset_type": "spreadsheet", "title": "Metrics Dashboard"},{"asset_type": "document", "title": "Executive Summary"},]batch_result = manager.create_batch_with_progress(batch_requests,on_progress=lambda completed, total: print(f"📊 Progress: {completed}/{total} assets created"))print("\n=== Batch Results ===")print(f"✅ Created: {batch_result.total_created}")print(f"❌ Failed: {batch_result.total_failed}")if batch_result.failed:print("\nFailed assets:")for item in batch_result.failed:print(f" - {item['request']['title']}: {item['error']}")except Exception as e:print(f"💥 Production example failed: {str(e)}")raise
Validation and Best Practices
Implement validation for asset creation:
VALID_ASSET_TYPES: List[AssetType] = ["spreadsheet","document","folder",]def validate_asset_type(asset_type: str) -> None:"""Validate that the asset type is supported.Args:asset_type: Asset type to validateRaises:ValueError: If asset type is not valid"""if asset_type not in VALID_ASSET_TYPES:raise ValueError(f"Invalid asset type: {asset_type}. "f"Valid types: {', '.join(VALID_ASSET_TYPES)}")def validate_title(title: str) -> None:"""Validate asset title.Args:title: Title to validateRaises:ValueError: If title is invalid"""if not title or not title.strip():raise ValueError("Asset title cannot be empty")if len(title) > 255:raise ValueError("Asset title cannot exceed 255 characters")def create_validated_asset(asset_type: str,title: str,parent_folder_id: Optional[str] = None) -> Dict[str, Any]:"""Create an asset with input validation.Args:asset_type: Type of asset to createtitle: Title for the assetparent_folder_id: Optional parent folder IDReturns:Created asset details"""# Validate inputsvalidate_asset_type(asset_type)validate_title(title)# If parent folder is specified, verify it exists (optional)if parent_folder_id:try:client.assets.get(parent_folder_id)except ApiError as e:if e.status_code == 404:raise ValueError(f"Parent folder not found: {parent_folder_id}")# If it's another error, continue anyway (might be permissions issue)# Create the assetresponse = client.assets.create(asset_type=asset_type,title=title,parent_folder_id=parent_folder_id)return {"asset_id": response.asset_id,"title": response.title,"asset_type": response.asset_type,"created_at": response.created_at,"parent_folder_id": response.parent_folder_id}
Workflow Integration Example
Integrate asset creation into a larger workflow:
@dataclassclass WorkflowConfig:"""Configuration for project workflow."""project_name: strdocument_count: intspreadsheet_count: intcreate_subfolders: bool = True@dataclassclass WorkflowResult:"""Result of workflow execution."""project_folder: AssetCreationResultdocuments: List[AssetCreationResult]spreadsheets: List[AssetCreationResult]subfolders: Optional[List[AssetCreationResult]] = Nonedef create_project_workflow(config: WorkflowConfig) -> WorkflowResult:"""Create a complete project workflow with assets.Args:config: Workflow configurationReturns:WorkflowResult with all created assets"""print(f"🚀 Starting project workflow: {config.project_name}")# Step 1: Create main project folderprint("📁 Step 1: Creating project folder...")project_folder = create_asset_with_retry("folder", config.project_name)if not project_folder.success:raise Exception(f"Failed to create project folder: {project_folder.error}")print(f"✅ Project folder created: {project_folder.asset_id}")# Step 2: Create subfolders if requestedsubfolders: Optional[List[AssetCreationResult]] = Noneif config.create_subfolders:print("📂 Step 2: Creating subfolders...")subfolder_requests = [{"asset_type": "folder", "title": "Documents", "parent_folder_id": project_folder.asset_id},{"asset_type": "folder", "title": "Spreadsheets", "parent_folder_id": project_folder.asset_id},{"asset_type": "folder", "title": "Reports", "parent_folder_id": project_folder.asset_id},]result = create_multiple_assets(subfolder_requests)subfolders = result.successfulprint(f"✅ Created {len(subfolders)} subfolders")# Step 3: Create documentsprint(f"📄 Step 3: Creating {config.document_count} documents...")document_requests = [{"asset_type": "document","title": f"Document {i + 1}","parent_folder_id": project_folder.asset_id}for i in range(config.document_count)]documents_result = create_multiple_assets(document_requests)# Step 4: Create spreadsheetsprint(f"📊 Step 4: Creating {config.spreadsheet_count} spreadsheets...")spreadsheet_requests = [{"asset_type": "spreadsheet","title": f"Spreadsheet {i + 1}","parent_folder_id": project_folder.asset_id}for i in range(config.spreadsheet_count)]spreadsheets_result = create_multiple_assets(spreadsheet_requests)print("\n=== Workflow Complete ===")print(f"Project Folder: {project_folder.asset_id}")print(f"Documents Created: {documents_result.total_created}")print(f"Spreadsheets Created: {spreadsheets_result.total_created}")if subfolders:print(f"Subfolders Created: {len(subfolders)}")return WorkflowResult(project_folder=project_folder,documents=documents_result.successful,spreadsheets=spreadsheets_result.successful,subfolders=subfolders)# Main executiondef main():"""Main function demonstrating complete workflow."""workflow_result = create_project_workflow(WorkflowConfig(project_name="Product Launch Q1 2024",document_count=5,spreadsheet_count=3,create_subfolders=True))print("🎉 Workflow completed successfully!")total_assets = (len(workflow_result.documents) +len(workflow_result.spreadsheets) +(len(workflow_result.subfolders) if workflow_result.subfolders else 0) +1 # project folder)print(f"Total assets created: {total_assets}")if __name__ == "__main__":main()
Async Asset Creation (Python 3.7+)
Use async/await for concurrent operations:
from athena import AsyncAthenaasync_client = AsyncAthena(api_key="YOUR_API_KEY")async def create_asset_async(asset_type: AssetType,title: str,parent_folder_id: Optional[str] = None) -> Dict[str, Any]:"""Create an asset asynchronously."""try:response = await async_client.assets.create(asset_type=asset_type,title=title,parent_folder_id=parent_folder_id)return {"asset_id": response.asset_id,"title": response.title,"asset_type": response.asset_type,"created_at": response.created_at,"success": True}except Exception as e:return {"title": title,"asset_type": asset_type,"success": False,"error": str(e)}async def create_multiple_assets_async(requests: List[Dict[str, Any]]) -> BatchAssetResult:"""Create multiple assets concurrently."""print(f"🔄 Creating {len(requests)} assets concurrently...")# Create all assets concurrentlytasks = [create_asset_async(req["asset_type"],req["title"],req.get("parent_folder_id"))for req in requests]results = await asyncio.gather(*tasks, return_exceptions=True)successful: List[AssetCreationResult] = []failed: List[dict] = []for i, result in enumerate(results):if isinstance(result, Exception):failed.append({"request": requests[i],"error": str(result)})elif isinstance(result, dict) and result.get("success"):successful.append(AssetCreationResult(asset_id=result["asset_id"],title=result["title"],asset_type=result["asset_type"],created_at=result["created_at"],success=True))else:failed.append({"request": requests[i],"error": result.get("error", "Unknown error")})return BatchAssetResult(successful=successful,failed=failed,total_created=len(successful),total_failed=len(failed))# Usageasync def async_example():"""Example using async asset creation."""requests = [{"asset_type": "spreadsheet", "title": f"Report {i}"}for i in range(10)]result = await create_multiple_assets_async(requests)print(f"Created {result.total_created} assets concurrently")# Run async exampleif __name__ == "__main__":asyncio.run(async_example())
Key Recommendations
- Use type hints - Define proper types for better code clarity
- Implement retry logic - Use exponential backoff for resilient operations
- Validate inputs - Check asset types and titles before API calls
- Handle errors gracefully - Distinguish between client and server errors
- Organize with folders - Create hierarchical structures for better organization
- Batch when possible - Create multiple assets efficiently
- Log progress - Provide visibility into long-running operations
- Use async for concurrency - Leverage AsyncAthena for parallel operations
Asset Types Supported: This endpoint currently supports three core asset types:
spreadsheet- Create Athena spreadsheets with real-time collaborationdocument- Create Athena documents for rich text editingfolder- Create folders for organizing your workspace
