0
# Uploads
1

2
Large file upload system with support for chunked uploads and multi-part file handling. Enables efficient upload of large files through resumable, chunked upload process with integrity verification.
3

4
## Capabilities
5

6
### Upload Management
7

8
Create, manage, and complete large file uploads using multi-part upload process for improved reliability and performance.
9

10
```python { .api }
11
class Uploads:
12
    def create(
13
        self,
14
        *,
15
        bytes: int,
16
        filename: str,
17
        mime_type: str,
18
        purpose: Any,
19
        **kwargs
20
    ) -> Upload:
21
        """
22
        Create a new upload session for large files.
23

24
        Args:
25
            bytes: Total size of the file in bytes
26
            filename: Name of the file being uploaded
27
            mime_type: MIME type of the file (e.g., "image/jpeg", "application/pdf")
28
            purpose: Purpose of the upload (e.g., "assistants", "fine-tune")
29
            **kwargs: Additional upload parameters
30

31
        Returns:
32
            Upload: Upload session object with upload ID and status
33
        """
34

35
    def upload_file_chunked(
36
        self,
37
        *,
38
        file: Union[os.PathLike[str], bytes],
39
        mime_type: str,
40
        purpose: Any,
41
        filename: Union[str, None] = None,
42
        bytes: Union[int, None] = None,
43
        part_size: Union[int, None] = None,
44
        md5: Union[str, NotGiven] = NOT_GIVEN
45
    ) -> Any:
46
        """
47
        Upload a file using automatic chunking for large files.
48

49
        Args:
50
            file: File path or file content bytes
51
            mime_type: MIME type of the file
52
            purpose: Purpose of the upload
53
            filename: Optional filename override
54
            bytes: Optional file size override
55
            part_size: Size of each chunk (default: 8MB)
56
            md5: MD5 hash for integrity verification
57

58
        Returns:
59
            File object after successful upload
60
        """
61

62
    def complete(
63
        self,
64
        upload_id: str,
65
        *,
66
        part_ids: List[str],
67
        md5: Union[str, NotGiven] = NOT_GIVEN,
68
        **kwargs
69
    ) -> Upload:
70
        """
71
        Complete a multi-part upload by assembling all parts.
72

73
        Args:
74
            upload_id: Upload session identifier
75
            part_ids: List of part IDs in order
76
            md5: MD5 hash of complete file for verification
77
            **kwargs: Additional completion parameters
78

79
        Returns:
80
            Upload: Completed upload object
81
        """
82

83
    def cancel(self, upload_id: str, **kwargs) -> Upload:
84
        """
85
        Cancel an in-progress upload session.
86

87
        Args:
88
            upload_id: Upload session identifier
89
            **kwargs: Additional cancellation parameters
90

91
        Returns:
92
            Upload: Cancelled upload object
93
        """
94

95
    parts: Parts
96

97
class AsyncUploads:
98
    async def create(
99
        self,
100
        *,
101
        bytes: int,
102
        filename: str,
103
        mime_type: str,
104
        purpose: Any,
105
        **kwargs
106
    ) -> Upload:
107
        """Async version of create method."""
108

109
    async def upload_file_chunked(
110
        self,
111
        *,
112
        file: Union[os.PathLike[str], bytes],
113
        mime_type: str,
114
        purpose: Any,
115
        filename: Union[str, None] = None,
116
        bytes: Union[int, None] = None,
117
        part_size: Union[int, None] = None,
118
        md5: Union[str, NotGiven] = NOT_GIVEN
119
    ) -> Any:
120
        """Async version of chunked upload method."""
121

122
    async def complete(
123
        self,
124
        upload_id: str,
125
        *,
126
        part_ids: List[str],
127
        md5: Union[str, NotGiven] = NOT_GIVEN,
128
        **kwargs
129
    ) -> Upload:
130
        """Async version of complete method."""
131

132
    async def cancel(self, upload_id: str, **kwargs) -> Upload:
133
        """Async version of cancel method."""
134

135
    parts: AsyncParts
136
```
137

138
### Part Management
139

140
Handle individual parts of multi-part uploads for fine-grained control over the upload process.
141

142
```python { .api }
143
class Parts:
144
    def create(
145
        self,
146
        upload_id: str,
147
        *,
148
        data: FileTypes,
149
        **kwargs
150
    ) -> UploadPart:
151
        """
152
        Upload a single part of a multi-part upload.
153

154
        Args:
155
            upload_id: Upload session identifier
156
            data: Part data as file-like object or bytes
157
            **kwargs: Additional part parameters
158

159
        Returns:
160
            UploadPart: Part object with part ID and status
161
        """
162

163
class AsyncParts:
164
    async def create(
165
        self,
166
        upload_id: str,
167
        *,
168
        data: FileTypes,
169
        **kwargs
170
    ) -> UploadPart:
171
        """Async version of part creation."""
172
```
173

174
### Usage Examples
175

176
```python
177
import os
178
from portkey_ai import Portkey
179

180
# Initialize client
181
portkey = Portkey(
182
    api_key="PORTKEY_API_KEY",
183
    virtual_key="VIRTUAL_KEY"
184
)
185

186
# Simple chunked upload (recommended for most use cases)
187
with open("large_document.pdf", "rb") as f:
188
    uploaded_file = portkey.uploads.upload_file_chunked(
189
        file=f,
190
        mime_type="application/pdf",
191
        purpose="assistants",
192
        filename="large_document.pdf"
193
    )
194

195
print(f"File uploaded successfully: {uploaded_file.id}")
196

197
# Manual multi-part upload for advanced use cases
198
file_path = "very_large_file.zip"
199
file_size = os.path.getsize(file_path)
200

201
# Create upload session
202
upload = portkey.uploads.create(
203
    bytes=file_size,
204
    filename="very_large_file.zip",
205
    mime_type="application/zip",
206
    purpose="assistants"
207
)
208

209
print(f"Upload session created: {upload.id}")
210

211
# Upload parts manually (8MB chunks)
212
part_size = 8 * 1024 * 1024  # 8MB
213
part_ids = []
214

215
with open(file_path, "rb") as f:
216
    part_number = 1
217
    while True:
218
        chunk = f.read(part_size)
219
        if not chunk:
220
            break
221
            
222
        part = portkey.uploads.parts.create(
223
            upload_id=upload.id,
224
            data=chunk
225
        )
226
        part_ids.append(part.id)
227
        print(f"Uploaded part {part_number}: {part.id}")
228
        part_number += 1
229

230
# Complete the upload
231
completed_upload = portkey.uploads.complete(
232
    upload_id=upload.id,
233
    part_ids=part_ids
234
)
235

236
print(f"Upload completed: {completed_upload.status}")
237
print(f"File ID: {completed_upload.file.id}")
238
```
239

240
### Async Usage
241

242
```python
243
import asyncio
244
import os
245
from portkey_ai import AsyncPortkey
246

247
async def upload_large_file():
248
    portkey = AsyncPortkey(
249
        api_key="PORTKEY_API_KEY",
250
        virtual_key="VIRTUAL_KEY"
251
    )
252
    
253
    # Async chunked upload
254
    with open("large_dataset.csv", "rb") as f:
255
        uploaded_file = await portkey.uploads.upload_file_chunked(
256
            file=f,
257
            mime_type="text/csv",
258
            purpose="fine-tune",
259
            part_size=16 * 1024 * 1024  # 16MB chunks
260
        )
261
    
262
    return uploaded_file
263

264
# Run async upload
265
uploaded_file = asyncio.run(upload_large_file())
266
print(f"Async upload completed: {uploaded_file.id}")
267
```
268

269
### Error Handling and Resumption
270

271
```python
272
import hashlib
273
import os
274

275
def upload_with_verification():
276
    file_path = "important_document.pdf"
277
    
278
    # Calculate MD5 for integrity verification
279
    with open(file_path, "rb") as f:
280
        file_hash = hashlib.md5(f.read()).hexdigest()
281
    
282
    file_size = os.path.getsize(file_path)
283
    
284
    try:
285
        # Create upload with MD5
286
        upload = portkey.uploads.create(
287
            bytes=file_size,
288
            filename=os.path.basename(file_path),
289
            mime_type="application/pdf",
290
            purpose="assistants"
291
        )
292
        
293
        # Upload with chunked method including hash
294
        with open(file_path, "rb") as f:
295
            uploaded_file = portkey.uploads.upload_file_chunked(
296
                file=f,
297
                mime_type="application/pdf",
298
                purpose="assistants",
299
                md5=file_hash
300
            )
301
        
302
        print(f"Upload successful with verification: {uploaded_file.id}")
303
        return uploaded_file
304
        
305
    except Exception as e:
306
        print(f"Upload failed: {e}")
307
        # Cancel the upload session if it was created
308
        if 'upload' in locals():
309
            try:
310
                cancelled = portkey.uploads.cancel(upload.id)
311
                print(f"Upload cancelled: {cancelled.status}")
312
            except Exception as cancel_error:
313
                print(f"Failed to cancel upload: {cancel_error}")
314
        
315
        raise
316

317
# Example with retry logic
318
def upload_with_retry(file_path, max_retries=3):
319
    for attempt in range(max_retries):
320
        try:
321
            return upload_with_verification()
322
        except Exception as e:
323
            print(f"Attempt {attempt + 1} failed: {e}")
324
            if attempt == max_retries - 1:
325
                raise
326
            # Wait before retry
327
            import time
328
            time.sleep(2 ** attempt)
329
```
330

331
## Types
332

333
```python { .api }
334
class Upload:
335
    """Upload session object"""
336
    id: str  # Upload identifier
337
    object: str  # "upload"
338
    bytes: int  # Total file size
339
    created_at: int  # Unix timestamp
340
    filename: str  # Original filename
341
    purpose: str  # Upload purpose
342
    status: str  # Upload status ("pending", "completed", "cancelled", "failed")
343
    expires_at: int  # Session expiration timestamp
344
    file: Optional[FileObject]  # File object after completion
345
    _headers: Optional[dict]  # Response headers
346

347
class UploadPart:
348
    """Individual upload part"""
349
    id: str  # Part identifier
350
    object: str  # "upload.part"
351
    created_at: int  # Unix timestamp
352
    upload_id: str  # Parent upload identifier
353
    etag: str  # Part ETag for integrity
354
    _headers: Optional[dict]  # Response headers
355

356
class FileObject:
357
    """Completed file object"""
358
    id: str  # File identifier
359
    object: str  # "file"
360
    bytes: int  # File size
361
    created_at: int  # Unix timestamp
362
    filename: str  # File name
363
    purpose: str  # File purpose
364
    status: str  # Processing status
365
    status_details: Optional[str]  # Status details if applicable
366

367
FileTypes = Union[
368
    bytes,
369
    str,
370
    os.PathLike[str],
371
    typing.IO[bytes],
372
    typing.IO[str]
373
]
374
"""Supported file input types for uploads"""
375
```