Performance Tuning

# ❌ Bad: Fetches all fields
data = dc.read_data("Company", output="pandas")
 
# ✅ Good: Fetches only needed fields
data = dc.read_data(
    "Company",
    output="pandas",
    fields=["CompanyName", "Revenue", "Status"]
)

# ❌ Bad: Fetch all, filter locally
data = dc.read_data("Company", output="pandas")
active = data[data["Status"] == "Active"]
 
# ✅ Good: Server-side filter
active = dc.read_data(
    "Company",
    output="pandas",
    query="{Status: 'Active'}"
)

# Bad: load everything
data = dc.read_data("LargeObject", output="list")
 
# Good: process batch by batch
for batch in dc.read_data_streaming("LargeObject", fields=["EntryId", "Name"]):
    for row in batch:
        process(row)

import tracemalloc
 
# Standard read
tracemalloc.start()
data = dc.read_data("Company", output="list")
current, peak = tracemalloc.get_traced_memory()
print(f"Standard: {peak / 1024 / 1024:.1f} MB")
tracemalloc.stop()
 
# Streaming read (each iteration is a batch)
tracemalloc.start()
for batch in dc.read_data_streaming("Company"):
    pass
current, peak = tracemalloc.get_traced_memory()
print(f"Streaming: {peak / 1024 / 1024:.1f} MB")
tracemalloc.stop()

# ❌ Bad: One API call per record
for record in records:
    dc.insert_data("Company", [record])
 
# ✅ Good: One API call for all records
dc.insert_data("Company", records)

def batch_insert(dc, object_id, records, batch_size=500):
    """Insert records in optimal batches."""
    results = []
    
    for i in range(0, len(records), batch_size):
        batch = records[i:i + batch_size]
        result = dc.insert_data(object_id, batch)
        results.extend(result)
        print(f"Inserted {len(results)}/{len(records)}")
    
    return results

# ❌ Bad: Send all fields
dc.update_data("Company", [
    {"EntryId": 123, "Name": "...", "Industry": "...", "Revenue": 1000}  # Unchanged fields
])
 
# ✅ Good: Send only changed fields
dc.update_data("Company", [
    {"EntryId": 123, "Revenue": 1000}  # Only the change
])

# ❌ Bad: Repeated schema calls
for obj in objects:
    schema = dc.get_schema()  # API call each iteration
    process(obj, schema)
 
# ✅ Good: Cache once
schema = dc.get_schema()
for obj in objects:
    process(obj, schema)

class FieldCache:
    def __init__(self, dc):
        self._cache = {}
        self._dc = dc
    
    def get_field(self, object_id, field_name):
        key = f"{object_id}.{field_name}"
        if key not in self._cache:
            fields = self._dc.get_fields(object_id)
            for f in fields:
                self._cache[f"{object_id}.{f.apiName}"] = f
        return self._cache.get(key)
 
# Usage
cache = FieldCache(dc)
field = cache.get_field("Company", "Revenue")  # First call: API
field = cache.get_field("Company", "Status")   # Cache hit!

from dealcloud_sdk import ReferenceCache
 
# Create shared cache
cache = ReferenceCache(max_entries=10000)
 
# Use across multiple reads
for obj in ["Contact", "Deal", "Interaction"]:
    data = dc.read_data(
        obj,
        output="pandas",
        reference_format=ReferenceFormat.NAME,
        reference_cache=cache  # Shared cache
    )

from dealcloud_sdk import DealCloudConfig, ConcurrencyLimits
 
config = DealCloudConfig(
    siteUrl="...",
    clientId=12345,
    clientSecret="...",
    concurrencyLimits=ConcurrencyLimits(read=4, create=2, delete=2),
)
 
dc = DealCloud.from_config_object(config)

# Reads pages in parallel up to read_concurrency (default 4)
data = dc.read_data("LargeObject", output="polars")
 
# Cap parallelism for this streaming read
for batch in dc.read_data_streaming("LargeObject", max_concurrency=2):
    process(batch)

import asyncio
 
async def export_one_object(dc, obj_name):
    rows = []
    async for batch in dc.aread_data_streaming(obj_name):
        rows.extend(batch)
    return obj_name, rows
 
async def parallel_export(dc, objects):
    return dict(await asyncio.gather(*[export_one_object(dc, o) for o in objects]))
 
# asyncio.run(parallel_export(dc, ["Company", "Contact", "Deal"]))

config = DealCloudConfig(
    # ...credentials
    querySettings={
        "pageSize": 2000,  # Larger pages, fewer requests
    }
)

config = DealCloudConfig(
    # ...credentials
    connectorTimeoutSeconds=180  # 3 minutes for large operations
)

import time
 
class PerformanceMonitor:
    def __init__(self):
        self.metrics = {}
    
    def time_operation(self, name):
        """Context manager to time operations."""
        class Timer:
            def __init__(self, monitor, name):
                self.monitor = monitor
                self.name = name
            
            def __enter__(self):
                self.start = time.time()
                return self
            
            def __exit__(self, *args):
                elapsed = time.time() - self.start
                self.monitor.metrics[self.name] = elapsed
                print(f"{self.name}: {elapsed:.2f}s")
        
        return Timer(self, name)
    
    def report(self):
        total = sum(self.metrics.values())
        print(f"\nTotal time: {total:.2f}s")
        for name, elapsed in sorted(self.metrics.items(), key=lambda x: -x[1]):
            pct = elapsed / total * 100
            print(f"  {name}: {elapsed:.2f}s ({pct:.1f}%)")
 
# Usage
monitor = PerformanceMonitor()
 
with monitor.time_operation("read_companies"):
    companies = dc.read_data("Company", output="list")
 
with monitor.time_operation("read_contacts"):
    contacts = dc.read_data("Contact", output="list")
 
monitor.report()