Upload files
Start with the SDK quickstart and configure FLASHDRIVE_API_KEY and FLASHDRIVE_DRIVE_ID. Use a key with write access. The examples below upload local files into that drive.
Upload a document
Save a file named handbook.md beside your script, then run:
import os
from flashdrive import Flashdrive
with Flashdrive(drive_id=os.environ["FLASHDRIVE_DRIVE_ID"]) as client:
batch = client.upload("handbook.md", route="uploads")
for receipt in batch.files:
print(receipt.filename, receipt.document_id, receipt.status)
print("Upload ID:", receipt.upload_id)The call waits for storage processing by default. Keep the document ID for later reads and updates. A stored file and a searchable file are different states; indexing can continue after storage finishes. Check processing status before relying on a search result.
The uploads route gives you per-file receipts. To upload several files, pass a list of paths. concurrency limits how many transfers run together:
import os
from flashdrive import Flashdrive
with Flashdrive(drive_id=os.environ["FLASHDRIVE_DRIVE_ID"]) as client:
batch = client.upload(
["handbook.md", "policy.txt"],
route="uploads",
concurrency=3,
timeout=600,
processing_timeout=600,
)
print("Stored receipts:", len(batch.files))Timeout values are in seconds. timeout bounds the upload phase; processing_timeout bounds the subsequent wait. With the default route="auto", large eligible text batches can use bulk import instead. In that case, read batch.import_status; batch.files is empty.
Select files carefully
Folder discovery follows .gitignore. A directly named file bypasses that ignore rule, but the SDK refuses explicit .git, .env, and system-leftover paths. Review the intended paths and keep credentials out of shared content. Hidden project files such as .gitignore, .github, and .env.example can be included. These discovery rules differ from mounts, which do not apply .gitignore.
Return before processing finishes
Use wait=False to return receipts after transfer and queueing. Save the upload and drive IDs in your application’s job record:
import os
from flashdrive import Flashdrive
with Flashdrive(drive_id=os.environ["FLASHDRIVE_DRIVE_ID"]) as client:
batch = client.upload("handbook.md", route="uploads", wait=False)
for receipt in batch.files:
print(receipt.upload_id, receipt.drive_id, receipt.status)Continue waiting later by passing those IDs. This resumes status polling, not an unfinished byte transfer:
import os
from flashdrive import Flashdrive
upload_id = input("Saved upload ID: ").strip()
with Flashdrive() as client:
receipt = client.uploads.resume(
upload_id,
drive_id=os.environ["FLASHDRIVE_DRIVE_ID"],
timeout=600,
)
print(receipt.document_id, receipt.status)
if receipt.operation_id:
operation = client.operations.get(
receipt.operation_id, drive_id=receipt.drive_id
)
print("Processing:", operation.status, operation.error_code)An operation can be queued, running, searchable, or failed. On failure, inspect its error code. If a wait times out, keep the receipt and continue polling instead of starting a duplicate upload.
Recover a partial batch
An upload batch can store some files while others fail. UploadBatchError retains both outcomes:
import os
from uuid import uuid4
from flashdrive import Flashdrive, UploadBatchError
operation_key = str(uuid4())
with Flashdrive(drive_id=os.environ["FLASHDRIVE_DRIVE_ID"]) as client:
try:
batch = client.upload(
["handbook.md", "policy.txt"],
route="uploads",
idempotency_key=operation_key,
)
print("Completed:", len(batch.files))
except UploadBatchError as error:
print("Completed:", len(error.batch.files))
for file_label, failure in error.failures.items():
print(file_label, type(failure).__name__)
print("Keep this operation key:", error.idempotency_key)
raisePersist the operation key before a recoverable job begins. A retry of the same upload must keep that key, file contents, and file order. A new upload job needs a new key. If an error contains a queued upload receipt, resume that receipt first. See errors and retries before building an automatic retry loop.