Create Assets

1
from athena import Athena
2
from athena.core import ApiError
3
4
# Production client setup
5
client = Athena(api_key="YOUR_API_KEY")
6
7
# Custom API endpoint (if needed)
8
custom_client = Athena(
9
    api_key="YOUR_API_KEY",
10
    base_url="https://your-custom-api.example.com"
11
)
12
13
# Local development
14
dev_client = Athena(
15
    api_key="YOUR_API_KEY",
16
    base_url="http://localhost:8000"
17
)

1
from typing import TypedDict, Optional, List, Literal
2
from dataclasses import dataclass
3
4
# Supported asset types
5
AssetType = Literal[
6
    "spreadsheet",
7
    "document", 
8
    "folder"
9
]
10
11
# Asset creation result
12
@dataclass
13
class AssetCreationResult:
14
    asset_id: str
15
    title: str
16
    asset_type: str
17
    created_at: str
18
    parent_folder_id: Optional[str] = None
19
    success: bool = True
20
    error: Optional[str] = None
21
22
# Batch creation result
23
@dataclass
24
class BatchAssetResult:
25
    successful: List[AssetCreationResult]
26
    failed: List[dict]
27
    total_created: int
28
    total_failed: int

1
def create_asset(
2
    asset_type: AssetType,
3
    title: Optional[str] = None,
4
    parent_folder_id: Optional[str] = None
5
) -> AssetCreationResult:
6
    """
7
    Create a single asset with error handling.
8
    
9
    Args:
10
        asset_type: Type of asset to create (spreadsheet, document, folder, etc.)
11
        title: Optional title for the asset
12
        parent_folder_id: Optional parent folder to create the asset in
13
        
14
    Returns:
15
        AssetCreationResult with success status and asset details
16
    """
17
    try:
18
        print(f"📝 Creating {asset_type}...")
19
        
20
        response = client.assets.create(
21
            asset_type=asset_type,
22
            title=title,
23
            parent_folder_id=parent_folder_id
24
        )
25
        
26
        print("✅ Asset created successfully")
27
        print(f"Asset ID: {response.asset_id}")
28
        print(f"Title: {response.title}")
29
        print(f"Created at: {response.created_at}")
30
        
31
        return AssetCreationResult(
32
            asset_id=response.asset_id,
33
            title=response.title,
34
            asset_type=response.asset_type,
35
            created_at=response.created_at,
36
            parent_folder_id=response.parent_folder_id,
37
            success=True
38
        )
39
    except ApiError as e:
40
        error_message = f"API Error ({e.status_code}): {str(e)}"
41
        print(f"❌ Asset creation failed: {error_message}")
42
        raise Exception(error_message)
43
    except Exception as e:
44
        error_message = f"Asset creation failed: {str(e)}"
45
        print(f"❌ {error_message}")
46
        raise Exception(error_message)
47
48
49
# Usage examples
50
def basic_examples():
51
    """Basic asset creation examples."""
52
    
53
    # Create a spreadsheet
54
    spreadsheet = create_asset("spreadsheet", "Q1 2024 Sales Report")
55
    
56
    # Create a document
57
    document = create_asset("document", "Meeting Notes")
58
    
59
    # Create a folder
60
    folder = create_asset("folder", "Project Alpha")
61
    
62
    return {
63
        "spreadsheet": spreadsheet,
64
        "document": document,
65
        "folder": folder
66
    }

1
def create_asset_in_folder(
2
    asset_type: AssetType,
3
    title: str,
4
    parent_folder_id: str
5
) -> AssetCreationResult:
6
    """
7
    Create an asset inside a specific folder.
8
    
9
    Args:
10
        asset_type: Type of asset to create
11
        title: Title for the asset
12
        parent_folder_id: ID of the parent folder
13
        
14
    Returns:
15
        AssetCreationResult with creation details
16
    """
17
    try:
18
        response = client.assets.create(
19
            asset_type=asset_type,
20
            title=title,
21
            parent_folder_id=parent_folder_id
22
        )
23
        
24
        print(f'✅ Created {asset_type} "{title}" in folder {parent_folder_id}')
25
        
26
        return AssetCreationResult(
27
            asset_id=response.asset_id,
28
            title=response.title,
29
            asset_type=response.asset_type,
30
            created_at=response.created_at,
31
            parent_folder_id=response.parent_folder_id,
32
            success=True
33
        )
34
    except Exception as e:
35
        error_message = str(e)
36
        return AssetCreationResult(
37
            asset_id="",
38
            title=title,
39
            asset_type=asset_type,
40
            created_at="",
41
            parent_folder_id=parent_folder_id,
42
            success=False,
43
            error=error_message
44
        )
45
46
47
def create_project_structure(project_name: str) -> dict:
48
    """
49
    Create an organized folder structure for a project.
50
    
51
    Args:
52
        project_name: Name of the project
53
        
54
    Returns:
55
        Dictionary with project folder and subfolders
56
    """
57
    print(f'📁 Creating project structure for "{project_name}"...')
58
    
59
    # Create main project folder
60
    project_folder = create_asset("folder", project_name)
61
    
62
    if not project_folder.success:
63
        raise Exception(f"Failed to create project folder: {project_folder.error}")
64
    
65
    print(f"✅ Created project folder: {project_folder.asset_id}")
66
    
67
    # Create subfolders
68
    subfolders_to_create = [
69
        {"name": "Documents", "type": "folder"},
70
        {"name": "Spreadsheets", "type": "folder"},
71
        {"name": "Reports", "type": "folder"},
72
    ]
73
    
74
    subfolders = [
75
        create_asset_in_folder(
76
            subfolder["type"],
77
            subfolder["name"],
78
            project_folder.asset_id
79
        )
80
        for subfolder in subfolders_to_create
81
    ]
82
    
83
    successful_subfolders = [f for f in subfolders if f.success]
84
    print(f"✅ Created {len(successful_subfolders)} subfolders")
85
    
86
    return {
87
        "project_folder": project_folder,
88
        "subfolders": subfolders
89
    }

1
from typing import List
2
import asyncio
3
from concurrent.futures import ThreadPoolExecutor
4
5
def create_multiple_assets(
6
    requests: List[dict]
7
) -> BatchAssetResult:
8
    """
9
    Create multiple assets with parallel execution.
10
    
11
    Args:
12
        requests: List of dicts with asset_type, title, and optional parent_folder_id
13
        
14
    Returns:
15
        BatchAssetResult with successful and failed creations
16
    """
17
    print(f"🔄 Creating {len(requests)} assets...")
18
    
19
    successful: List[AssetCreationResult] = []
20
    failed: List[dict] = []
21
    
22
    for req in requests:
23
        try:
24
            response = client.assets.create(
25
                asset_type=req["asset_type"],
26
                title=req.get("title"),
27
                parent_folder_id=req.get("parent_folder_id")
28
            )
29
            
30
            successful.append(AssetCreationResult(
31
                asset_id=response.asset_id,
32
                title=response.title,
33
                asset_type=response.asset_type,
34
                created_at=response.created_at,
35
                parent_folder_id=response.parent_folder_id,
36
                success=True
37
            ))
38
        except Exception as e:
39
            failed.append({
40
                "request": req,
41
                "error": str(e)
42
            })
43
    
44
    print(f"✅ Successfully created: {len(successful)}")
45
    print(f"❌ Failed: {len(failed)}")
46
    
47
    return BatchAssetResult(
48
        successful=successful,
49
        failed=failed,
50
        total_created=len(successful),
51
        total_failed=len(failed)
52
    )
53
54
55
def create_quarterly_reports():
56
    """Example: Create multiple quarterly reports."""
57
    quarters = ["Q1", "Q2", "Q3", "Q4"]
58
    year = "2024"
59
    
60
    requests = [
61
        {
62
            "asset_type": "spreadsheet",
63
            "title": f"{quarter} {year} Sales Report"
64
        }
65
        for quarter in quarters
66
    ]
67
    
68
    result = create_multiple_assets(requests)
69
    
70
    print("\n=== Batch Creation Results ===")
71
    print(f"Total created: {result.total_created}")
72
    print(f"Total failed: {result.total_failed}")
73
    
74
    if result.failed:
75
        print("\nFailed creations:")
76
        for item in result.failed:
77
            print(f"  - {item['request']['title']}: {item['error']}")
78
    
79
    return result

1
from typing import Dict, Any
2
3
class FolderStructure(TypedDict, total=False):
4
    name: str
5
    type: AssetType
6
    children: List['FolderStructure']
7
8
class CreatedAssetNode(TypedDict, total=False):
9
    asset_id: str
10
    title: str
11
    asset_type: str
12
    children: List['CreatedAssetNode']
13
14
15
def create_folder_hierarchy(
16
    structure: FolderStructure,
17
    parent_folder_id: Optional[str] = None
18
) -> CreatedAssetNode:
19
    """
20
    Recursively create a folder hierarchy from a structure definition.
21
    
22
    Args:
23
        structure: Folder structure definition
24
        parent_folder_id: Optional parent folder ID
25
        
26
    Returns:
27
        CreatedAssetNode with created asset details
28
    """
29
    # Create the current folder/asset
30
    response = client.assets.create(
31
        asset_type=structure["type"],
32
        title=structure["name"],
33
        parent_folder_id=parent_folder_id
34
    )
35
    
36
    print(f'✅ Created {structure["type"]}: {structure["name"]}')
37
    
38
    node: CreatedAssetNode = {
39
        "asset_id": response.asset_id,
40
        "title": response.title,
41
        "asset_type": response.asset_type
42
    }
43
    
44
    # Recursively create children
45
    if "children" in structure and structure["children"]:
46
        child_nodes = [
47
            create_folder_hierarchy(child, response.asset_id)
48
            for child in structure["children"]
49
        ]
50
        node["children"] = child_nodes
51
    
52
    return node
53
54
55
def create_complete_project_structure():
56
    """Example: Create a complete project structure."""
57
    project_structure: FolderStructure = {
58
        "name": "Product Launch 2024",
59
        "type": "folder",
60
        "children": [
61
            {
62
                "name": "Research",
63
                "type": "folder",
64
                "children": [
65
                    {"name": "Market Analysis", "type": "spreadsheet"},
66
                    {"name": "Competitor Research", "type": "document"},
67
                    {"name": "User Survey Results", "type": "spreadsheet"},
68
                ]
69
            },
70
            {
71
                "name": "Planning",
72
                "type": "folder",
73
                "children": [
74
                    {"name": "Timeline", "type": "spreadsheet"},
75
                    {"name": "Budget", "type": "spreadsheet"},
76
                    {"name": "Strategy Doc", "type": "document"},
77
                ]
78
            },
79
            {
80
                "name": "Deliverables",
81
                "type": "folder",
82
                "children": [
83
                    {"name": "Launch Plan", "type": "document"},
84
                    {"name": "Marketing Materials", "type": "folder"},
85
                ]
86
            }
87
        ]
88
    }
89
    
90
    print("🏗️  Creating complete project structure...")
91
    result = create_folder_hierarchy(project_structure)
92
    print("🎉 Project structure created successfully!")
93
    
94
    return result

1
import time
2
from typing import Optional
3
4
def create_asset_with_retry(
5
    asset_type: AssetType,
6
    title: str,
7
    parent_folder_id: Optional[str] = None,
8
    max_retries: int = 3
9
) -> AssetCreationResult:
10
    """
11
    Create an asset with automatic retry logic.
12
    
13
    Args:
14
        asset_type: Type of asset to create
15
        title: Title for the asset
16
        parent_folder_id: Optional parent folder ID
17
        max_retries: Maximum number of retry attempts
18
        
19
    Returns:
20
        AssetCreationResult with creation details
21
    """
22
    last_error: Optional[Exception] = None
23
    
24
    for attempt in range(1, max_retries + 1):
25
        try:
26
            print(f"🔄 Attempt {attempt}/{max_retries} to create {asset_type}")
27
            
28
            response = client.assets.create(
29
                asset_type=asset_type,
30
                title=title,
31
                parent_folder_id=parent_folder_id
32
            )
33
            
34
            print(f"✅ Asset created successfully on attempt {attempt}")
35
            
36
            return AssetCreationResult(
37
                asset_id=response.asset_id,
38
                title=response.title,
39
                asset_type=response.asset_type,
40
                created_at=response.created_at,
41
                parent_folder_id=response.parent_folder_id,
42
                success=True
43
            )
44
        except ApiError as e:
45
            last_error = e
46
            
47
            # Don't retry on client errors (4xx)
48
            if 400 <= e.status_code < 500:
49
                print(f"❌ Client error ({e.status_code}): {str(e)}")
50
                return AssetCreationResult(
51
                    asset_id="",
52
                    title=title,
53
                    asset_type=asset_type,
54
                    created_at="",
55
                    parent_folder_id=parent_folder_id,
56
                    success=False,
57
                    error=f"Client error ({e.status_code}): {str(e)}"
58
                )
59
            
60
            # Retry on server errors or network issues
61
            if attempt < max_retries:
62
                delay = min(1.0 * (2 ** (attempt - 1)), 10.0)
63
                print(f"⏳ Waiting {delay}s before retry...")
64
                time.sleep(delay)
65
        except Exception as e:
66
            last_error = e
67
            
68
            if attempt < max_retries:
69
                delay = min(1.0 * (2 ** (attempt - 1)), 10.0)
70
                print(f"⏳ Waiting {delay}s before retry...")
71
                time.sleep(delay)
72
    
73
    print("❌ Asset creation failed after all retry attempts")
74
    error_message = str(last_error) if last_error else "Unknown error"
75
    
76
    return AssetCreationResult(
77
        asset_id="",
78
        title=title,
79
        asset_type=asset_type,
80
        created_at="",
81
        parent_folder_id=parent_folder_id,
82
        success=False,
83
        error=f"Failed after {max_retries} attempts: {error_message}"
84
    )

1
from typing import Dict, Any
2
3
class AssetFactory:
4
    """Factory class for creating different types of assets."""
5
    
6
    def __init__(self, api_key: str, base_url: Optional[str] = None):
7
        """
8
        Initialize the asset factory.
9
        
10
        Args:
11
            api_key: Athena API key
12
            base_url: Optional custom base URL
13
        """
14
        self.client = Athena(api_key=api_key, base_url=base_url)
15
    
16
    def create_spreadsheet(
17
        self,
18
        title: str,
19
        parent_folder_id: Optional[str] = None
20
    ) -> Dict[str, Any]:
21
        """Create a spreadsheet asset."""
22
        return self._create_asset("spreadsheet", title, parent_folder_id)
23
    
24
    def create_document(
25
        self,
26
        title: str,
27
        parent_folder_id: Optional[str] = None
28
    ) -> Dict[str, Any]:
29
        """Create a document asset."""
30
        return self._create_asset("document", title, parent_folder_id)
31
    
32
    def create_folder(
33
        self,
34
        title: str,
35
        parent_folder_id: Optional[str] = None
36
    ) -> Dict[str, Any]:
37
        """Create a folder asset."""
38
        return self._create_asset("folder", title, parent_folder_id)
39
    
40
    def _create_asset(
41
        self,
42
        asset_type: AssetType,
43
        title: str,
44
        parent_folder_id: Optional[str] = None
45
    ) -> Dict[str, Any]:
46
        """Internal method to create an asset."""
47
        try:
48
            response = self.client.assets.create(
49
                asset_type=asset_type,
50
                title=title,
51
                parent_folder_id=parent_folder_id
52
            )
53
            
54
            print(f'✅ Created {asset_type}: {response.title} ({response.asset_id})')
55
            
56
            return {
57
                "asset_id": response.asset_id,
58
                "title": response.title,
59
                "asset_type": response.asset_type,
60
                "created_at": response.created_at,
61
                "parent_folder_id": response.parent_folder_id
62
            }
63
        except ApiError as e:
64
            raise Exception(
65
                f"Failed to create {asset_type}: {e.status_code} - {str(e)}"
66
            )
67
    
68
    def create_batch(
69
        self,
70
        requests: List[Dict[str, Any]]
71
    ) -> BatchAssetResult:
72
        """
73
        Create multiple assets in batch.
74
        
75
        Args:
76
            requests: List of asset creation requests
77
            
78
        Returns:
79
            BatchAssetResult with successful and failed creations
80
        """
81
        print(f"🔄 Creating {len(requests)} assets in batch...")
82
        
83
        successful: List[AssetCreationResult] = []
84
        failed: List[dict] = []
85
        
86
        for req in requests:
87
            try:
88
                response = self._create_asset(
89
                    req["type"],
90
                    req["title"],
91
                    req.get("parent_folder_id")
92
                )
93
                
94
                successful.append(AssetCreationResult(
95
                    asset_id=response["asset_id"],
96
                    title=response["title"],
97
                    asset_type=response["asset_type"],
98
                    created_at=response["created_at"],
99
                    parent_folder_id=response.get("parent_folder_id"),
100
                    success=True
101
                ))
102
            except Exception as e:
103
                failed.append({
104
                    "request": req,
105
                    "error": str(e)
106
                })
107
        
108
        return BatchAssetResult(
109
            successful=successful,
110
            failed=failed,
111
            total_created=len(successful),
112
            total_failed=len(failed)
113
        )
114
115
116
# Usage
117
def factory_example():
118
    """Example using the asset factory."""
119
    factory = AssetFactory(api_key="YOUR_API_KEY")
120
    
121
    # Create individual assets
122
    spreadsheet = factory.create_spreadsheet("Sales Dashboard")
123
    document = factory.create_document("Product Requirements")
124
    folder = factory.create_folder("Marketing Campaign")
125
    
126
    # Create batch of assets
127
    batch_result = factory.create_batch([
128
        {"type": "spreadsheet", "title": "Budget 2024"},
129
        {"type": "document", "title": "Project Brief"},
130
        {"type": "folder", "title": "Assets"},
131
    ])
132
    
133
    print(f"Created {batch_result.total_created} assets")
134
    
135
    return {
136
        "individual": [spreadsheet, document, folder],
137
        "batch": batch_result
138
    }

1
import os
2
from typing import Optional, List, Callable
3
from dataclasses import dataclass, field
4
5
@dataclass
6
class AssetCreationOptions:
7
    """Options for asset creation."""
8
    retries: int = 3
9
    timeout: int = 60
10
    throw_on_error: bool = False
11
    parent_folder_id: Optional[str] = None
12
13
14
class AssetManager:
15
    """Production-ready asset manager with comprehensive error handling."""
16
    
17
    def __init__(self, api_key: str, base_url: Optional[str] = None):
18
        """
19
        Initialize the asset manager.
20
        
21
        Args:
22
            api_key: Athena API key
23
            base_url: Optional custom base URL
24
        """
25
        self.client = Athena(api_key=api_key, base_url=base_url)
26
        self.default_retries = 3
27
    
28
    def create(
29
        self,
30
        asset_type: AssetType,
31
        title: str,
32
        options: Optional[AssetCreationOptions] = None
33
    ) -> AssetCreationResult:
34
        """
35
        Create an asset with comprehensive error handling.
36
        
37
        Args:
38
            asset_type: Type of asset to create
39
            title: Title for the asset
40
            options: Creation options including retries and parent folder
41
            
42
        Returns:
43
            AssetCreationResult with creation details
44
        """
45
        if options is None:
46
            options = AssetCreationOptions()
47
        
48
        retries = options.retries or self.default_retries
49
        parent_folder_id = options.parent_folder_id
50
        throw_on_error = options.throw_on_error
51
        
52
        attempt = 0
53
        last_error: Optional[Exception] = None
54
        
55
        while attempt < retries:
56
            try:
57
                attempt += 1
58
                print(f"📝 Creating {asset_type} (attempt {attempt}/{retries})...")
59
                
60
                response = self.client.assets.create(
61
                    asset_type=asset_type,
62
                    title=title,
63
                    parent_folder_id=parent_folder_id
64
                )
65
                
66
                print(f"✅ Asset created: {response.asset_id}")
67
                
68
                return AssetCreationResult(
69
                    asset_id=response.asset_id,
70
                    title=response.title,
71
                    asset_type=response.asset_type,
72
                    created_at=response.created_at,
73
                    parent_folder_id=response.parent_folder_id,
74
                    success=True
75
                )
76
            except ApiError as e:
77
                last_error = e
78
                
79
                # Don't retry on client errors
80
                if 400 <= e.status_code < 500:
81
                    print(f"❌ Client error ({e.status_code}): {str(e)}")
82
                    
83
                    if throw_on_error:
84
                        raise
85
                    
86
                    return AssetCreationResult(
87
                        asset_id="",
88
                        title=title,
89
                        asset_type=asset_type,
90
                        created_at="",
91
                        parent_folder_id=parent_folder_id,
92
                        success=False,
93
                        error=f"Client error ({e.status_code}): {str(e)}"
94
                    )
95
                
96
                print(f"⚠️  Attempt {attempt} failed: {str(e)}")
97
                
98
                # Wait before retry with exponential backoff
99
                if attempt < retries:
100
                    delay = min(1.0 * (2 ** (attempt - 1)), 10.0)
101
                    time.sleep(delay)
102
            except Exception as e:
103
                last_error = e
104
                print(f"⚠️  Attempt {attempt} failed: {str(e)}")
105
                
106
                if attempt < retries:
107
                    delay = min(1.0 * (2 ** (attempt - 1)), 10.0)
108
                    time.sleep(delay)
109
        
110
        error_message = str(last_error) if last_error else "Unknown error"
111
        
112
        if throw_on_error:
113
            raise Exception(f"Asset creation failed after {retries} attempts: {error_message}")
114
        
115
        return AssetCreationResult(
116
            asset_id="",
117
            title=title,
118
            asset_type=asset_type,
119
            created_at="",
120
            parent_folder_id=parent_folder_id,
121
            success=False,
122
            error=f"Failed after {retries} attempts: {error_message}"
123
        )
124
    
125
    def create_from_template(
126
        self,
127
        template: FolderStructure,
128
        parent_folder_id: Optional[str] = None
129
    ) -> CreatedAssetNode:
130
        """
131
        Create a workspace structure from a template.
132
        
133
        Args:
134
            template: Folder structure template
135
            parent_folder_id: Optional parent folder ID
136
            
137
        Returns:
138
            CreatedAssetNode with created structure
139
        """
140
        print(f'🏗️  Creating structure from template: {template["name"]}')
141
        
142
        response = self.create(
143
            template["type"],
144
            template["name"],
145
            AssetCreationOptions(
146
                parent_folder_id=parent_folder_id,
147
                throw_on_error=True
148
            )
149
        )
150
        
151
        node: CreatedAssetNode = {
152
            "asset_id": response.asset_id,
153
            "title": response.title,
154
            "asset_type": response.asset_type
155
        }
156
        
157
        if "children" in template and template["children"]:
158
            print(f'📂 Creating {len(template["children"])} children...')
159
            
160
            child_nodes = [
161
                self.create_from_template(child, response.asset_id)
162
                for child in template["children"]
163
            ]
164
            
165
            node["children"] = child_nodes
166
        
167
        return node
168
    
169
    def create_batch_with_progress(
170
        self,
171
        requests: List[Dict[str, Any]],
172
        on_progress: Optional[Callable[[int, int], None]] = None
173
    ) -> BatchAssetResult:
174
        """
175
        Create multiple assets with progress tracking.
176
        
177
        Args:
178
            requests: List of asset creation requests
179
            on_progress: Optional callback for progress updates
180
            
181
        Returns:
182
            BatchAssetResult with successful and failed creations
183
        """
184
        total = len(requests)
185
        print(f"🔄 Creating {total} assets with progress tracking...")
186
        
187
        successful: List[AssetCreationResult] = []
188
        failed: List[dict] = []
189
        
190
        for i, req in enumerate(requests):
191
            try:
192
                response = self.client.assets.create(
193
                    asset_type=req["asset_type"],
194
                    title=req.get("title"),
195
                    parent_folder_id=req.get("parent_folder_id")
196
                )
197
                
198
                successful.append(AssetCreationResult(
199
                    asset_id=response.asset_id,
200
                    title=response.title,
201
                    asset_type=response.asset_type,
202
                    created_at=response.created_at,
203
                    parent_folder_id=response.parent_folder_id,
204
                    success=True
205
                ))
206
                
207
                if on_progress:
208
                    on_progress(len(successful), total)
209
            except Exception as e:
210
                failed.append({
211
                    "request": req,
212
                    "error": str(e)
213
                })
214
        
215
        return BatchAssetResult(
216
            successful=successful,
217
            failed=failed,
218
            total_created=len(successful),
219
            total_failed=len(failed)
220
        )
221
222
223
def production_example():
224
    """Complete production usage example."""
225
    manager = AssetManager(api_key=os.getenv("ATHENA_API_KEY"))
226
    
227
    try:
228
        # Create a single asset with retry
229
        document = manager.create(
230
            "document",
231
            "Important Document",
232
            AssetCreationOptions(retries=5)
233
        )
234
        
235
        if document.success:
236
            print(f"✅ Document created: {document.asset_id}")
237
        
238
        # Create from template
239
        project_template: FolderStructure = {
240
            "name": "Q1 2024 Initiative",
241
            "type": "folder",
242
            "children": [
243
                {
244
                    "name": "Planning",
245
                    "type": "folder",
246
                    "children": [
247
                        {"name": "Project Plan", "type": "document"},
248
                        {"name": "Budget Tracker", "type": "spreadsheet"},
249
                    ]
250
                },
251
                {
252
                    "name": "Execution",
253
                    "type": "folder",
254
                    "children": [
255
                        {"name": "Tasks", "type": "spreadsheet"},
256
                        {"name": "Progress Report", "type": "document"},
257
                    ]
258
                }
259
            ]
260
        }
261
        
262
        project = manager.create_from_template(project_template)
263
        print(f"✅ Project structure created: {project['asset_id']}")
264
        
265
        # Batch creation with progress
266
        batch_requests = [
267
            {"asset_type": "spreadsheet", "title": "Sales Data"},
268
            {"asset_type": "document", "title": "Analysis Report"},
269
            {"asset_type": "spreadsheet", "title": "Metrics Dashboard"},
270
            {"asset_type": "document", "title": "Executive Summary"},
271
        ]
272
        
273
        batch_result = manager.create_batch_with_progress(
274
            batch_requests,
275
            on_progress=lambda completed, total: print(f"📊 Progress: {completed}/{total} assets created")
276
        )
277
        
278
        print("\n=== Batch Results ===")
279
        print(f"✅ Created: {batch_result.total_created}")
280
        print(f"❌ Failed: {batch_result.total_failed}")
281
        
282
        if batch_result.failed:
283
            print("\nFailed assets:")
284
            for item in batch_result.failed:
285
                print(f"  - {item['request']['title']}: {item['error']}")
286
    except Exception as e:
287
        print(f"💥 Production example failed: {str(e)}")
288
        raise

1
VALID_ASSET_TYPES: List[AssetType] = [
2
    "spreadsheet",
3
    "document",
4
    "folder",
5
]
6
7
def validate_asset_type(asset_type: str) -> None:
8
    """
9
    Validate that the asset type is supported.
10
    
11
    Args:
12
        asset_type: Asset type to validate
13
        
14
    Raises:
15
        ValueError: If asset type is not valid
16
    """
17
    if asset_type not in VALID_ASSET_TYPES:
18
        raise ValueError(
19
            f"Invalid asset type: {asset_type}. "
20
            f"Valid types: {', '.join(VALID_ASSET_TYPES)}"
21
        )
22
23
def validate_title(title: str) -> None:
24
    """
25
    Validate asset title.
26
    
27
    Args:
28
        title: Title to validate
29
        
30
    Raises:
31
        ValueError: If title is invalid
32
    """
33
    if not title or not title.strip():
34
        raise ValueError("Asset title cannot be empty")
35
    
36
    if len(title) > 255:
37
        raise ValueError("Asset title cannot exceed 255 characters")
38
39
40
def create_validated_asset(
41
    asset_type: str,
42
    title: str,
43
    parent_folder_id: Optional[str] = None
44
) -> Dict[str, Any]:
45
    """
46
    Create an asset with input validation.
47
    
48
    Args:
49
        asset_type: Type of asset to create
50
        title: Title for the asset
51
        parent_folder_id: Optional parent folder ID
52
        
53
    Returns:
54
        Created asset details
55
    """
56
    # Validate inputs
57
    validate_asset_type(asset_type)
58
    validate_title(title)
59
    
60
    # If parent folder is specified, verify it exists (optional)
61
    if parent_folder_id:
62
        try:
63
            client.assets.get(parent_folder_id)
64
        except ApiError as e:
65
            if e.status_code == 404:
66
                raise ValueError(f"Parent folder not found: {parent_folder_id}")
67
            # If it's another error, continue anyway (might be permissions issue)
68
    
69
    # Create the asset
70
    response = client.assets.create(
71
        asset_type=asset_type,
72
        title=title,
73
        parent_folder_id=parent_folder_id
74
    )
75
    
76
    return {
77
        "asset_id": response.asset_id,
78
        "title": response.title,
79
        "asset_type": response.asset_type,
80
        "created_at": response.created_at,
81
        "parent_folder_id": response.parent_folder_id
82
    }

1
@dataclass
2
class WorkflowConfig:
3
    """Configuration for project workflow."""
4
    project_name: str
5
    document_count: int
6
    spreadsheet_count: int
7
    create_subfolders: bool = True
8
9
10
@dataclass
11
class WorkflowResult:
12
    """Result of workflow execution."""
13
    project_folder: AssetCreationResult
14
    documents: List[AssetCreationResult]
15
    spreadsheets: List[AssetCreationResult]
16
    subfolders: Optional[List[AssetCreationResult]] = None
17
18
19
def create_project_workflow(config: WorkflowConfig) -> WorkflowResult:
20
    """
21
    Create a complete project workflow with assets.
22
    
23
    Args:
24
        config: Workflow configuration
25
        
26
    Returns:
27
        WorkflowResult with all created assets
28
    """
29
    print(f"🚀 Starting project workflow: {config.project_name}")
30
    
31
    # Step 1: Create main project folder
32
    print("📁 Step 1: Creating project folder...")
33
    project_folder = create_asset_with_retry("folder", config.project_name)
34
    
35
    if not project_folder.success:
36
        raise Exception(f"Failed to create project folder: {project_folder.error}")
37
    
38
    print(f"✅ Project folder created: {project_folder.asset_id}")
39
    
40
    # Step 2: Create subfolders if requested
41
    subfolders: Optional[List[AssetCreationResult]] = None
42
    if config.create_subfolders:
43
        print("📂 Step 2: Creating subfolders...")
44
        subfolder_requests = [
45
            {"asset_type": "folder", "title": "Documents", "parent_folder_id": project_folder.asset_id},
46
            {"asset_type": "folder", "title": "Spreadsheets", "parent_folder_id": project_folder.asset_id},
47
            {"asset_type": "folder", "title": "Reports", "parent_folder_id": project_folder.asset_id},
48
        ]
49
        
50
        result = create_multiple_assets(subfolder_requests)
51
        subfolders = result.successful
52
        print(f"✅ Created {len(subfolders)} subfolders")
53
    
54
    # Step 3: Create documents
55
    print(f"📄 Step 3: Creating {config.document_count} documents...")
56
    document_requests = [
57
        {
58
            "asset_type": "document",
59
            "title": f"Document {i + 1}",
60
            "parent_folder_id": project_folder.asset_id
61
        }
62
        for i in range(config.document_count)
63
    ]
64
    
65
    documents_result = create_multiple_assets(document_requests)
66
    
67
    # Step 4: Create spreadsheets
68
    print(f"📊 Step 4: Creating {config.spreadsheet_count} spreadsheets...")
69
    spreadsheet_requests = [
70
        {
71
            "asset_type": "spreadsheet",
72
            "title": f"Spreadsheet {i + 1}",
73
            "parent_folder_id": project_folder.asset_id
74
        }
75
        for i in range(config.spreadsheet_count)
76
    ]
77
    
78
    spreadsheets_result = create_multiple_assets(spreadsheet_requests)
79
    
80
    print("\n=== Workflow Complete ===")
81
    print(f"Project Folder: {project_folder.asset_id}")
82
    print(f"Documents Created: {documents_result.total_created}")
83
    print(f"Spreadsheets Created: {spreadsheets_result.total_created}")
84
    if subfolders:
85
        print(f"Subfolders Created: {len(subfolders)}")
86
    
87
    return WorkflowResult(
88
        project_folder=project_folder,
89
        documents=documents_result.successful,
90
        spreadsheets=spreadsheets_result.successful,
91
        subfolders=subfolders
92
    )
93
94
95
# Main execution
96
def main():
97
    """Main function demonstrating complete workflow."""
98
    workflow_result = create_project_workflow(
99
        WorkflowConfig(
100
            project_name="Product Launch Q1 2024",
101
            document_count=5,
102
            spreadsheet_count=3,
103
            create_subfolders=True
104
        )
105
    )
106
    
107
    print("🎉 Workflow completed successfully!")
108
    total_assets = (
109
        len(workflow_result.documents) +
110
        len(workflow_result.spreadsheets) +
111
        (len(workflow_result.subfolders) if workflow_result.subfolders else 0) +
112
        1  # project folder
113
    )
114
    print(f"Total assets created: {total_assets}")
115
116
117
if __name__ == "__main__":
118
    main()

1
from athena import AsyncAthena
2
3
async_client = AsyncAthena(api_key="YOUR_API_KEY")
4
5
async def create_asset_async(
6
    asset_type: AssetType,
7
    title: str,
8
    parent_folder_id: Optional[str] = None
9
) -> Dict[str, Any]:
10
    """Create an asset asynchronously."""
11
    try:
12
        response = await async_client.assets.create(
13
            asset_type=asset_type,
14
            title=title,
15
            parent_folder_id=parent_folder_id
16
        )
17
        
18
        return {
19
            "asset_id": response.asset_id,
20
            "title": response.title,
21
            "asset_type": response.asset_type,
22
            "created_at": response.created_at,
23
            "success": True
24
        }
25
    except Exception as e:
26
        return {
27
            "title": title,
28
            "asset_type": asset_type,
29
            "success": False,
30
            "error": str(e)
31
        }
32
33
34
async def create_multiple_assets_async(
35
    requests: List[Dict[str, Any]]
36
) -> BatchAssetResult:
37
    """Create multiple assets concurrently."""
38
    print(f"🔄 Creating {len(requests)} assets concurrently...")
39
    
40
    # Create all assets concurrently
41
    tasks = [
42
        create_asset_async(
43
            req["asset_type"],
44
            req["title"],
45
            req.get("parent_folder_id")
46
        )
47
        for req in requests
48
    ]
49
    
50
    results = await asyncio.gather(*tasks, return_exceptions=True)
51
    
52
    successful: List[AssetCreationResult] = []
53
    failed: List[dict] = []
54
    
55
    for i, result in enumerate(results):
56
        if isinstance(result, Exception):
57
            failed.append({
58
                "request": requests[i],
59
                "error": str(result)
60
            })
61
        elif isinstance(result, dict) and result.get("success"):
62
            successful.append(AssetCreationResult(
63
                asset_id=result["asset_id"],
64
                title=result["title"],
65
                asset_type=result["asset_type"],
66
                created_at=result["created_at"],
67
                success=True
68
            ))
69
        else:
70
            failed.append({
71
                "request": requests[i],
72
                "error": result.get("error", "Unknown error")
73
            })
74
    
75
    return BatchAssetResult(
76
        successful=successful,
77
        failed=failed,
78
        total_created=len(successful),
79
        total_failed=len(failed)
80
    )
81
82
83
# Usage
84
async def async_example():
85
    """Example using async asset creation."""
86
    requests = [
87
        {"asset_type": "spreadsheet", "title": f"Report {i}"}
88
        for i in range(10)
89
    ]
90
    
91
    result = await create_multiple_assets_async(requests)
92
    print(f"Created {result.total_created} assets concurrently")
93
94
95
# Run async example
96
if __name__ == "__main__":
97
    asyncio.run(async_example())