Typed Data with Pydantic
The SDK supports type-safe data operations using Pydantic models, providing IDE autocomplete, runtime validation, and cleaner code. Typed helpers call the same HTTP layer as untyped methods (intapp-rest-client).
Overview
đź’ˇ
Key concept: You define Pydantic models in your code to match your DealCloud schema. The SDK provides generic typed methods that work with any model.
Benefits
| Feature | Description |
|---|---|
| IDE Autocomplete | Full property suggestions as you type |
| Type Checking | Catch type errors before runtime |
| Validation | Automatic data validation on read/write |
| Cleaner Code | company.Name instead of company["Name"] |
Defining Models
Create Pydantic models matching your DealCloud fields:
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
class Company(BaseModel):
EntryId: int
CompanyName: str
Industry: Optional[str] = None
Revenue: Optional[float] = None
Status: Optional[int] = None # Choice field (ID)
CreatedDate: Optional[datetime] = None
class Contact(BaseModel):
EntryId: int
FirstName: str
LastName: str
Email: Optional[str] = None
Company: Optional[int] = None # Reference field (EntryId)
@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 reference
CloseDate: Optional[datetime] = NoneModel Guidelines
| Field Type | Python Type | Notes |
|---|---|---|
| Text | str | Required or Optional[str] |
| Number | float or int | Use Optional for nullable |
| Date | datetime | Auto-parsed from ISO strings |
| Choice | int | Choice value ID |
| Reference | int | Entry ID |
| Multi-select | List[int] | List of IDs |
| User | int | User ID |
Reading Typed Data
typed_read_data()
from dealcloud_sdk import DealCloud, DealCloudConfig
config = DealCloudConfig(
siteUrl="yoursite.dealcloud.com",
clientId=12345,
clientSecret="your-secret",
)
dc = DealCloud.from_config_object(config)
# Read as typed models
companies = dc.typed_read_data(Company, object_id="Company")
for company in companies:
print(f"{company.CompanyName}: ${company.Revenue:,.0f}")
# Full autocomplete on company.CompanyName, company.Revenue, etc.With Query Filter
# Filter results
active = dc.typed_read_data(
Company,
object_id="Company",
query="{Status: 'Active'}"
)From View
# Read from configured view
deals = dc.typed_read_data(
Deal,
view_id="Active Deals"
)Field Auto-Detection
Fields are automatically inferred from the model:
# Model defines which fields to fetch
class CompanySummary(BaseModel):
EntryId: int
CompanyName: str
Revenue: Optional[float] = None
# Only fetches EntryId, CompanyName, Revenue
summaries = dc.typed_read_data(CompanySummary, object_id="Company")Or specify explicitly:
# Override auto-detection
companies = dc.typed_read_data(
Company,
object_id="Company",
fields=["CompanyName", "Industry"] # Only these fields
)Streaming Typed Data
For large datasets:
# Memory-efficient typed streaming
for company in dc.typed_read_data_streaming(Company, object_id="Company"):
process(company)Writing Typed Data
typed_insert_data()
# Create new records
new_companies = [
Company(CompanyName="Acme Corp", Industry="Technology", Revenue=1000000),
Company(CompanyName="Beta Inc", Industry="Finance", Revenue=500000),
]
# Insert and get back with EntryIds
inserted = dc.typed_insert_data("Company", new_companies, Company)
for company in inserted:
print(f"Created {company.CompanyName} with ID {company.EntryId}")đź’ˇ
For inserts, EntryId in your model should be optional or have a default. The API assigns IDs.
typed_update_data()
# Fetch, modify, update
companies = dc.typed_read_data(
Company,
object_id="Company",
query="{Industry: 'Tech'}"
)
# Modify
for company in companies:
company.Industry = "Technology" # Standardize
# Update
updated = dc.typed_update_data("Company", companies, Company)typed_upsert_data()
# Upsert with match field
class CompanySync(BaseModel):
EntryId: Optional[int] = None
ExternalId: str # Match on this
CompanyName: str
Revenue: Optional[float] = None
sync_data = [
CompanySync(ExternalId="CRM-001", CompanyName="Acme Corp", Revenue=1000000),
CompanySync(ExternalId="CRM-002", CompanyName="Beta Inc", Revenue=500000),
]
result = dc.typed_upsert_data(
"Company",
sync_data,
CompanySync,
match_field="ExternalId"
)Validation Handling
Strict Mode (Default)
Raises on validation errors:
try:
companies = dc.typed_read_data(Company, object_id="Company")
except ValidationError as e:
print(f"Validation failed: {e}")Lenient Mode
Skip invalid records:
companies = dc.typed_read_data(
Company,
object_id="Company",
skip_validation_errors=True # Log warnings, skip bad records
)Advanced Model Patterns
Computed Properties
class Contact(BaseModel):
EntryId: int
FirstName: str
LastName: str
Email: Optional[str] = None
@property
def full_name(self) -> str:
return f"{self.FirstName} {self.LastName}"
@property
def email_domain(self) -> Optional[str]:
if self.Email and "@" in self.Email:
return self.Email.split("@")[1]
return None
contacts = dc.typed_read_data(Contact, object_id="Contact")
for c in contacts:
print(f"{c.full_name} ({c.email_domain})")Field Aliases
from pydantic import Field
class Company(BaseModel):
EntryId: int
name: str = Field(alias="CompanyName") # API uses CompanyName
revenue: Optional[float] = Field(alias="AnnualRevenue")
class Config:
populate_by_name = True # Accept both name and aliasCustom Validators
from pydantic import field_validator
class Deal(BaseModel):
EntryId: int
DealName: str
Value: Optional[float] = None
@field_validator("Value")
@classmethod
def value_must_be_positive(cls, v):
if v is not None and v < 0:
raise ValueError("Value must be positive")
return vModel for Write vs Read
# Read model (includes system fields)
class CompanyRead(BaseModel):
EntryId: int
CompanyName: str
CreatedDate: datetime
ModifiedDate: datetime
# Write model (no system fields)
class CompanyWrite(BaseModel):
CompanyName: str
Industry: Optional[str] = None
Revenue: Optional[float] = None
# Use appropriate model for each operation
existing = dc.typed_read_data(CompanyRead, object_id="Company")
dc.typed_insert_data("Company", [CompanyWrite(CompanyName="New Corp")], CompanyWrite)Parameters
typed_read_data()
| Parameter | Type | Description |
|---|---|---|
model | Type[T] | Pydantic model class |
object_id | str | int | Object API name or ID |
view_id | str | int | View (alternative to object_id) |
fields | List[str] | Fields to fetch (default: from model) |
query | str | Filter query |
skip_validation_errors | bool | Skip invalid rows with warning |
typed_insert_data() / typed_update_data() / typed_upsert_data()
| Parameter | Type | Description |
|---|---|---|
object_id | str | int | Object API name or ID |
data | List[T] | List of model instances |
model | Type[T] | Model class for return type |
match_field | str | Match field for upsert |
Related
- Basic Reads - Standard reads
- Streaming - Memory-efficient reads
- Insert Data - Untyped inserts
- Data models - SDK dataclasses vs your Pydantic models
- Utilities & helpers - Profiling and file ingest helpers