Data API
Read Operations
Typed Data (Pydantic)

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

FeatureDescription
IDE AutocompleteFull property suggestions as you type
Type CheckingCatch type errors before runtime
ValidationAutomatic data validation on read/write
Cleaner Codecompany.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] = None

Model Guidelines

Field TypePython TypeNotes
TextstrRequired or Optional[str]
Numberfloat or intUse Optional for nullable
DatedatetimeAuto-parsed from ISO strings
ChoiceintChoice value ID
ReferenceintEntry ID
Multi-selectList[int]List of IDs
UserintUser 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 alias

Custom 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 v

Model 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()

ParameterTypeDescription
modelType[T]Pydantic model class
object_idstr | intObject API name or ID
view_idstr | intView (alternative to object_id)
fieldsList[str]Fields to fetch (default: from model)
querystrFilter query
skip_validation_errorsboolSkip invalid rows with warning

typed_insert_data() / typed_update_data() / typed_upsert_data()

ParameterTypeDescription
object_idstr | intObject API name or ID
dataList[T]List of model instances
modelType[T]Model class for return type
match_fieldstrMatch field for upsert

Related