Bulk File Export
Export images and documents from DealCloud in bulk with support for local storage and custom backends. Exports use read_data() plus download_file() internally; HTTP is via intapp-rest-client.
Overview
| Method | Exports | Field Type |
|---|---|---|
export_images() | Image fields | fieldType: 16 |
export_documents() | Document attachments | fieldType: 13 |
export_files() | Both images and documents | Both |
export_entry_files() | All files from one entry | Both |
export_images()
Export image fields from DealCloud:
from dealcloud_sdk import DealCloud, DealCloudConfig
config = DealCloudConfig(
siteUrl="yoursite.dealcloud.com",
clientId=12345,
clientSecret="your-secret",
)
dc = DealCloud.from_config_object(config)
# Export all images to local directory
result = dc.export_images("./backup/images")
print(f"Total: {result.total_files}")
print(f"Exported: {result.exported}")
print(f"Failed: {result.failed}")Scoped Export
# All images from all objects
result = dc.export_images("./backup/all_images")With Query Filter
# Export logos from active companies only
result = dc.export_images(
"./backup/active_logos",
object_id="Company",
field_ids=["Logo"],
query="{Status: 'Active'}"
)export_documents()
Export document/binary attachments:
# Export all documents
result = dc.export_documents("./backup/documents")
# Export from specific object
result = dc.export_documents(
"./backup/deal_docs",
object_id="Deal"
)
# Export specific attachment fields
result = dc.export_documents(
"./backup/contracts",
object_id="Deal",
field_ids=["Contract", "Proposal"]
)export_files()
Combined export of images and documents:
# Export everything
result = dc.export_files("./backup/all_files")
# Images only
result = dc.export_files(
"./backup/images_only",
include_images=True,
include_documents=False
)
# Documents only
result = dc.export_files(
"./backup/docs_only",
include_images=False,
include_documents=True
)export_entry_files()
Export all files from a single record:
# Export all files for a specific deal
result = dc.export_entry_files(
object_id="Deal",
entry_id=12345,
output_dir="./deal_12345_files"
)Parameters
Common Parameters
| Parameter | Type | Description |
|---|---|---|
output_dir | str | Path | Local directory for files |
storage | StorageBackend | Custom storage (overrides output_dir) |
object_id | str | int | Filter to specific object |
entry_ids | List[int] | Filter to specific entries |
field_ids | List[str | int] | Filter to specific fields |
query | str | DealCloud query filter |
naming | str | Path template |
overwrite | bool | Overwrite existing files |
on_progress | Callable | Progress callback |
Naming Template
Control output file paths with placeholders:
| Placeholder | Description |
|---|---|
{object} | Object API name |
{entry_id} | Entry ID |
{field_name} | Field API name |
{ext} | File extension |
{date} | Current date |
# Default naming
naming = "{object}/{entry_id}_{field_name}.{ext}"
# Output: Company/12345_Logo.png
# Flat structure
naming = "{object}_{entry_id}_{field_name}.{ext}"
# Output: Company_12345_Logo.png
# Date-organized
naming = "{date}/{object}/{entry_id}.{ext}"
# Output: 2024-01-15/Company/12345.pngReturn Value
@dataclass
class ExportResult:
total_files: int # Total files found
exported: int # Successfully exported
failed: int # Failed exports
skipped: int # Skipped (already exists)
errors: List[dict] # Error details
output_path: str # Base output pathProgress Tracking
def on_progress(current, total):
print(f"Exported {current}/{total} files ({current/total*100:.1f}%)")
result = dc.export_images(
"./backup/images",
on_progress=on_progress
)Custom Storage Backends
LocalStorage (Built-in)
from dealcloud_sdk import LocalStorage
# Explicit local storage
storage = LocalStorage("./backup/images")
result = dc.export_images(storage=storage)
# Network share (Windows UNC)
storage = LocalStorage(r"\\server\share\dealcloud\images")
result = dc.export_images(storage=storage)StorageBackend Protocol
Implement for custom storage:
from typing import Protocol
class StorageBackend(Protocol):
def write(self, path: str, content: bytes, content_type: str) -> str:
"""Write content to storage. Returns full path/URL."""
...
def exists(self, path: str) -> bool:
"""Check if file already exists."""
...Bulk export passes full file bytes to write(); that matches most cloud SDKs. For streaming writes to a URI (e.g. very large objects written chunk-by-chunk), use FsspecStreamingStorage or LocalStorage.open_write—see Data models (StreamingStorageBackend protocol).
S3 Storage Example
class S3Storage:
def __init__(self, bucket: str, prefix: str = ""):
import boto3
self.s3 = boto3.client('s3')
self.bucket = bucket
self.prefix = prefix
def write(self, path: str, content: bytes, content_type: str) -> str:
key = f"{self.prefix}/{path}" if self.prefix else path
self.s3.put_object(
Bucket=self.bucket,
Key=key,
Body=content,
ContentType=content_type
)
return f"s3://{self.bucket}/{key}"
def exists(self, path: str) -> bool:
key = f"{self.prefix}/{path}" if self.prefix else path
try:
self.s3.head_object(Bucket=self.bucket, Key=key)
return True
except:
return False
# Usage
s3 = S3Storage("my-bucket", "dealcloud/images")
result = dc.export_images(storage=s3, object_id="Company")Azure Blob Example
class AzureBlobStorage:
def __init__(self, connection_string: str, container: str):
from azure.storage.blob import BlobServiceClient
client = BlobServiceClient.from_connection_string(connection_string)
self.container = client.get_container_client(container)
def write(self, path: str, content: bytes, content_type: str) -> str:
blob = self.container.get_blob_client(path)
blob.upload_blob(content, content_type=content_type, overwrite=True)
return blob.url
def exists(self, path: str) -> bool:
return self.container.get_blob_client(path).exists()
# Usage
azure = AzureBlobStorage(conn_string, "dealcloud")
result = dc.export_documents(storage=azure)Complete Backup Example
from dealcloud_sdk import DealCloud, DealCloudConfig
from datetime import datetime
from pathlib import Path
def backup_all_files(dc, base_dir: str):
"""Complete file backup with logging."""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_dir = Path(base_dir) / timestamp
def log_progress(current, total):
print(f"\r Progress: {current}/{total}", end="", flush=True)
# Backup images
print("Exporting images...")
images = dc.export_images(
backup_dir / "images",
on_progress=log_progress
)
print(f"\n Images: {images.exported}/{images.total_files}")
# Backup documents
print("Exporting documents...")
docs = dc.export_documents(
backup_dir / "documents",
on_progress=log_progress
)
print(f"\n Documents: {docs.exported}/{docs.total_files}")
# Summary
total = images.exported + docs.exported
print(f"\nBackup complete: {total} files to {backup_dir}")
return {
"images": images,
"documents": docs,
"path": str(backup_dir)
}
config = DealCloudConfig(
siteUrl="yoursite.dealcloud.com",
clientId=12345,
clientSecret="your-secret",
)
dc = DealCloud.from_config_object(config)
result = backup_all_files(dc, "./backups")Related
- Download Files - Single file download
- Upload Files - Uploading files
- Backups API - Full site backups