Async Operations

import asyncio
from intapp_rest_client import RestClient, OAuth2Config
 
async def main():
    async with RestClient(
        base_url="https://api.example.com",
        oauth2_config=OAuth2Config(
            token_url="https://api.example.com/oauth/token",
            client_id="client_id",
            client_secret="secret"
        )
    ) as client:
        # All standard HTTP methods have async versions
        data = await client.aget("/api/v4/resources")
        result = await client.apost("/api/v4/resources", json={"name": "New"})
        await client.adelete("/api/v4/resources/123")
 
asyncio.run(main())

async with RestClient(base_url="https://api.example.com", api_key="key") as client:
    data = await client.aget("/api/resources")
# Client is automatically closed

client = RestClient(base_url="https://api.example.com", api_key="key")
try:
    data = await client.aget("/api/resources")
finally:
    await client.aclose()  # Don't forget to close!

import asyncio
from intapp_rest_client import RestClient
 
async def fetch_all_data(client):
    # Run multiple requests concurrently
    results = await asyncio.gather(
        client.aget("/api/users"),
        client.aget("/api/companies"),
        client.aget("/api/deals"),
    )
    users, companies, deals = results
    return {"users": users, "companies": companies, "deals": deals}
 
async def main():
    async with RestClient(base_url="https://api.example.com", api_key="key") as client:
        data = await fetch_all_data(client)
        print(f"Fetched {len(data['users'])} users")
 
asyncio.run(main())

async def fetch_with_fallback(client, endpoints):
    """Fetch from multiple endpoints, handling individual failures."""
    tasks = [client.aget(endpoint) for endpoint in endpoints]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    successful = []
    failed = []
    for endpoint, result in zip(endpoints, results):
        if isinstance(result, Exception):
            failed.append((endpoint, result))
        else:
            successful.append(result)
    
    return successful, failed

async def process_all_companies(client):
    async for batch in client.aget_paginated(
        "/api/v4/data/rows/Company",
        page_size=1000,
        data_key="rows"
    ):
        for company in batch:
            await process_company(company)
 
async def main():
    async with RestClient(base_url="https://api.example.com", api_key="key") as client:
        await process_all_companies(client)
 
asyncio.run(main())

# Pagination in request body
async for batch in client.apost_paginated(
    "/api/v4/query",
    page_size=1000,
    data_key="results",
    json={"filter": {"status": "Active"}}
):
    process_batch(batch)
 
# Pagination in query params (for APIs that use POST with URL pagination)
async for batch in client.apost_paginated_params(
    "/api/v4/data/rows/view/123",
    page_size=1000,
    data_key="rows",
    params={"wrapIntoArrays": "true"}
):
    process_batch(batch)

async def download_file(client, file_id, output_path):
    response = await client.aget_raw(f"/api/files/{file_id}/download")
    
    import aiofiles
    async with aiofiles.open(output_path, "wb") as f:
        await f.write(response.content)

async with RestClient(base_url="https://api.example.com", api_key="key") as client:
    size = await client.adownload_file("/api/files/123/download", "/tmp/output.pdf")

async def upload_file(client, file_path, destination):
    import aiofiles
    async with aiofiles.open(file_path, "rb") as f:
        content = await f.read()
    
    response = await client.apost_raw(
        "/api/files/upload",
        files={"file": (file_path.name, content, "application/octet-stream")}
    )
    return response.json()

async def fetch_with_limit(client, endpoints, max_concurrent=10):
    """Fetch with limited concurrency."""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def fetch_one(endpoint):
        async with semaphore:
            return await client.aget(endpoint)
    
    tasks = [fetch_one(endpoint) for endpoint in endpoints]
    return await asyncio.gather(*tasks)
 
async def main():
    async with RestClient(base_url="https://api.example.com", api_key="key") as client:
        endpoints = [f"/api/items/{i}" for i in range(100)]
        results = await fetch_with_limit(client, endpoints, max_concurrent=5)

client = RestClient(base_url="https://api.example.com", api_key="key")
 
# Sync usage
data = client.get("/api/resources")
 
# Async usage  
async def async_operation():
    return await client.aget("/api/resources")
 
asyncio.run(async_operation())
 
# Remember to close both
client.close()  # Closes sync client
asyncio.run(client.aclose())  # Closes async client

# ✅ Good
async with RestClient(...) as client:
    await client.aget("/api/data")
 
# ❌ Avoid - easy to forget cleanup
client = RestClient(...)
await client.aget("/api/data")
# Did you remember to call await client.aclose()?

# ✅ Good - controlled concurrency
semaphore = asyncio.Semaphore(10)
async with semaphore:
    await client.aget(endpoint)
 
# ❌ Avoid - unlimited concurrency can overwhelm APIs
await asyncio.gather(*[client.aget(e) for e in endpoints])  # 1000 concurrent!

# ✅ Good - handle individual failures
results = await asyncio.gather(*tasks, return_exceptions=True)
 
# ❌ Avoid - one failure cancels everything
results = await asyncio.gather(*tasks)  # Raises on first error

# ✅ Good - reuse client
async with RestClient(...) as client:
    for item in items:
        await client.apost("/api/items", json=item)
 
# ❌ Avoid - new client per request
for item in items:
    async with RestClient(...) as client:
        await client.apost("/api/items", json=item)