Async Operations

import asyncio
from dealcloud_sdk import DealCloud, DealCloudConfig
 
async def main():
    config = DealCloudConfig(
        siteUrl="yoursite.dealcloud.com",
        clientId=12345,
        clientSecret="your-secret",
    )
    dc = DealCloud.from_config_object(config)
 
    # Async streaming read
    async for company in dc.aread_data_streaming("Company"):
        await process_company(company)
 
asyncio.run(main())

from pydantic import BaseModel
 
class Company(BaseModel):
    EntryId: int
    CompanyName: str
 
async def typed_stream(dc):
    async for company in dc.typed_aread_data_streaming(Company, object_id="Company"):
        print(company.CompanyName)  # Full type hints!

async def sync_all_objects(dc):
    """Sync multiple objects concurrently."""
    
    objects = ["Company", "Contact", "Deal"]
    
    async def sync_object(obj_name):
        count = 0
        async for record in dc.aread_data_streaming(obj_name):
            await save_to_db(obj_name, record)
            count += 1
        return (obj_name, count)
    
    results = await asyncio.gather(*[sync_object(obj) for obj in objects])
    
    for obj_name, count in results:
        print(f"{obj_name}: {count} records synced")

from datetime import datetime, timedelta
 
async def async_delta_sync(dc, object_id: str, last_sync: datetime):
    """Async delta synchronization."""
    
    # Get modified entries (async)
    changes = await dc.aget_modified_entries(object_id, last_sync)
    
    modified_ids = [c.entry_id for c in changes if not c.is_deleted]
    deleted_ids = [c.entry_id for c in changes if c.is_deleted]
    
    # Process modifications
    if modified_ids:
        ids_query = f"{{EntryId: {{$in: [{','.join(map(str, modified_ids))}]}}}}"
        async for record in dc.aread_data_streaming(object_id, query=ids_query):
            await upsert_to_db(record)
    
    # Process deletions
    for entry_id in deleted_ids:
        await delete_from_db(entry_id)
    
    return {"modified": len(modified_ids), "deleted": len(deleted_ids)}

from fastapi import FastAPI
from dealcloud_sdk import DealCloud, DealCloudConfig
 
app = FastAPI()
config = DealCloudConfig(
    siteUrl="yoursite.dealcloud.com",
    clientId=12345,
    clientSecret="your-secret",
)
dc = DealCloud.from_config_object(config)
 
@app.get("/companies")
async def list_companies():
    companies = []
    async for company in dc.aread_data_streaming("Company"):
        companies.append(company)
        if len(companies) >= 100:  # Limit for API response
            break
    return companies
 
@app.get("/companies/{entry_id}")
async def get_company(entry_id: int):
    # Use sync method for single record
    result = dc.read_data(
        "Company",
        output="list",
        query=f"{{EntryId: {entry_id}}}"
    )
    return result[0] if result else {"error": "Not found"}

from dealcloud_sdk import DealCloud, DealCloudConfig
 
async def with_cleanup():
    config = DealCloudConfig(
        siteUrl="yoursite.dealcloud.com",
        clientId=12345,
        clientSecret="your-secret",
    )
    dc = DealCloud.from_config_object(config)
 
    try:
        async for record in dc.aread_data_streaming("Company"):
            await process(record)
    finally:
        # Cleanup if needed
        pass

async def robust_async_processing(dc, object_id: str):
    """Async processing with error handling."""
    
    errors = []
    processed = 0
    
    try:
        async for record in dc.aread_data_streaming(object_id):
            try:
                await process_record(record)
                processed += 1
            except Exception as e:
                errors.append({
                    "entry_id": record.get("EntryId"),
                    "error": str(e)
                })
    except asyncio.CancelledError:
        print("Operation cancelled")
        raise
    
    return {"processed": processed, "errors": errors}

import asyncio
 
class RateLimiter:
    def __init__(self, rate_per_second: float):
        self.rate = rate_per_second
        self.last_call = 0
    
    async def wait(self):
        now = asyncio.get_event_loop().time()
        elapsed = now - self.last_call
        wait_time = max(0, (1 / self.rate) - elapsed)
        if wait_time > 0:
            await asyncio.sleep(wait_time)
        self.last_call = asyncio.get_event_loop().time()
 
async def rate_limited_processing(dc):
    limiter = RateLimiter(10)  # 10 requests/second
    
    async for record in dc.aread_data_streaming("Company"):
        await limiter.wait()
        await api_call(record)