Data Models
Configuration and data-helper models used by the SDK. DealCloudConfig and related types are Pydantic-based; many schema types are dataclasses (Object, Field, …)—see dealcloud_sdk.models. For method-level return types, prefer the API reference.
For user-defined row types with Pydantic (typed_read_data, …), see Typed data. This page focuses on SDK-supplied configuration and helper types.
Configuration Models
DealCloudConfig
from dealcloud_sdk import DealCloudConfig
config = DealCloudConfig(
siteUrl="yoursite.dealcloud.com",
clientId=12345,
clientSecret="your-secret",
connectorTimeoutSeconds=100,
querySettings=QuerySettings(...),
concurrencyLimits=ConcurrencyLimits(...),
responseRetrySettings=ResponseRetrySettings(...)
)Properties:
| Property | Type | Required | Default |
|---|---|---|---|
siteUrl | str | Yes | - |
clientId | int | Yes | - |
clientSecret | str | Yes | - |
connectorTimeoutSeconds | int | No | 100 |
querySettings | QuerySettings | No | defaults |
concurrencyLimits | ConcurrencyLimits | No | defaults |
responseRetrySettings | ResponseRetrySettings | No | defaults |
QuerySettings
from dealcloud_sdk import QuerySettings
settings = QuerySettings(
pageSize=1000,
cellPaginationLimit=9000,
deletePageSize=10000
)ConcurrencyLimits
from dealcloud_sdk import ConcurrencyLimits
limits = ConcurrencyLimits(
read=2,
delete=2,
create=2
)ResponseRetrySettings
from dealcloud_sdk import ResponseRetrySettings
retry = ResponseRetrySettings(
tooManyRequests=5,
internalServerError=2,
serviceUnavailable=2,
backoffFactor=2.0
)Schema Models
Object
@dataclass
class Object:
id: int
name: str
apiName: str
singularName: str
pluralName: str
entryListType: int
entryListSubType: intField
@dataclass
class Field:
id: int
name: str
apiName: str
fieldType: int
isRequired: bool
isMultiSelect: bool
isCalculated: bool
isAttachment: bool
description: Optional[str]
formula: Optional[str]
choiceValues: Optional[List[ChoiceValue]]
entryLists: Optional[List[int]]
systemFieldType: Optional[int]ChoiceValue
@dataclass
class ChoiceValue:
id: int
name: str
parentID: Optional[int]
seqNumber: intUser
@dataclass
class User:
id: int
name: str
email: str
firstName: str
lastName: str
isActive: boolSchema
@dataclass
class Schema:
objects: Dict[str, ObjectSchema]
@dataclass
class ObjectSchema:
object: Object
fields: Dict[str, Field]Data Models
ModifiedEntry
from dealcloud_sdk import ModifiedEntry
@dataclass
class ModifiedEntry:
entry_id: int
modified_date: datetime
is_deleted: boolDeltaSyncResult
from dealcloud_sdk import DeltaSyncResult
@dataclass
class DeltaSyncResult:
sync_timestamp: datetime
modified_ids: List[int]
deleted_ids: List[int]
modified_data: Union[List[dict], pd.DataFrame, pl.DataFrame, pl.LazyFrame]The modified_data type depends on the output parameter passed to sync_delta():
output="list"→List[dict]output="pandas"→pd.DataFrameoutput="polars"→pl.DataFrameoutput="polars_lazy"→pl.LazyFrame
EntryIdCache
from dealcloud_sdk import EntryIdCache
cache = EntryIdCache(
object_id="Company",
key_field="ExternalId"
)
# Methods
cache.get(external_id) -> Optional[int]
cache.set(external_id, entry_id)
cache.bulk_set(mappings)
len(cache)
external_id in cacheReferenceCache
from dealcloud_sdk import ReferenceCache
cache = ReferenceCache(max_entries=10000)
# Methods
cache.get(object_id, entry_id) -> Optional[str]
cache.set(object_id, entry_id, name)
cache.bulk_set(object_id, mappings)
cache.get_missing(object_id, entry_ids) -> List[int]
len(cache)BatchResult
Aggregates successes and failures for batch-style operations when using ErrorHandling.COLLECT (or similar paths that return partial results). See Error handling.
Transport failures are in errors. Row-level Errors from HTTP 200 responses are in row_errors (each item is a full row dict with EntryId).
from dealcloud_sdk import BatchResult
# results: list — successful rows (no "Errors" key)
# errors: list — transport / parallel batch failures
# row_errors: list — rows with "Errors" from HTTP 200 body
# total_requested, total_succeeded, total_failed: int
def summarize(br: BatchResult) -> None:
if br.has_errors:
print(f"Success rate: {br.success_rate:.1f}%")
print(f"Row errors: {len(br.row_errors)}, transport errors: {len(br.errors)}")RowsWriteResult
Structured result when output="write_result" on insert_data / update_data / upsert_data / write_cells ( error_handling=FAIL_FAST only for Rows; Cells supports "list" or "write_result" only).
from dealcloud_sdk import RowsWriteResult
result: RowsWriteResult = dc.update_data("Company", records, output="write_result")
print(result.rows) # full API list (ok + error rows)
print(result.ok_rows) # rows without "Errors"
print(result.row_errors) # rows with "Errors"
print(result.has_row_errors)File Models
ExportResult
from dealcloud_sdk import ExportResult
@dataclass
class ExportResult:
total_files: int
exported: int
failed: int
skipped: int
errors: List[dict]
output_path: Optional[str]StorageBackend (Protocol)
from dealcloud_sdk import StorageBackend
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."""
...LocalStorage
from dealcloud_sdk import LocalStorage
storage = LocalStorage(base_dir="./exports")
# Methods
storage.write(path, content, content_type) -> str
storage.exists(path) -> boolStreamingStorageBackend (Protocol)
Long-lived streaming writes (e.g. large downloads) use a different protocol than buffer-then-write StorageBackend. Implementations expose open_write(path, *, content_type) as a context manager yielding a writable binary file-like object.
FsspecStreamingStorage
FsspecStreamingStorage implements streaming writes rooted at a URI (file://, s3://, abfs://, gs://, …). Install optional backends with the same extras as cloud file export: s3fs, adlfs, gcsfs (see installation).
from dealcloud_sdk import FsspecStreamingStorage
# Write large objects without holding full bytes in memory
storage = FsspecStreamingStorage("s3://my-bucket/prefix", **{"key": "...", "secret": "..."})
with storage.open_write("export.bin", content_type="application/octet-stream") as f:
f.write(chunk)Use LocalStorage for local paths when you need both write() (bytes) and open_write() (streaming). For attachment-style exports that pass full bytes, StorageBackend / LocalStorage.write are enough—see Bulk export.
Enums
ReferenceFormat
from dealcloud_sdk import ReferenceFormat
class ReferenceFormat(Enum):
ID = "id" # Just entry IDs
NAME = "name" # Just display names
FULL = "full" # Full reference objectErrorHandling
from dealcloud_sdk import ErrorHandling
ErrorHandling.FAIL_FAST # Stop on first error (default)
ErrorHandling.COLLECT # Continue, collect errors
ErrorHandling.LOG_ONLY # Log errors, continueCreating Custom Models
For typed data operations, define your own Pydantic models:
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
class Company(BaseModel):
EntryId: int
CompanyName: str
Industry: Optional[int] = None
Revenue: Optional[float] = None
Status: Optional[int] = None
CreatedDate: Optional[datetime] = None
class Config:
# Allow field names from API
populate_by_name = True
class Contact(BaseModel):
EntryId: int
FirstName: str
LastName: str
Email: Optional[str] = None
Company: Optional[int] = None # Reference as ID
@property
def full_name(self) -> str:
return f"{self.FirstName} {self.LastName}"
class Deal(BaseModel):
EntryId: int
DealName: str
Value: Optional[float] = None
Stage: Optional[int] = None
Companies: Optional[List[int]] = None # Multi-select