Boost Performance: Async Python for Didit API & Webhooks

Optimize API InteractionsLeverage Python's asyncio to make non-blocking calls to Didit's API, drastically improving throughput for identity verification and reducing latency for high-volume operations.

Efficient Webhook ProcessingImplement asynchronous webhook handlers to process incoming Didit events without blocking, ensuring your application remains responsive and can scale to handle numerous concurrent notifications.

Master Rate Limit ManagementUnderstand and effectively manage Didit's API rate limits using asynchronous techniques and intelligent backoff strategies, preventing service interruptions and ensuring reliable data exchange.

Didit's AI-Native AdvantageDidit's platform is designed for high performance and developer-friendliness, offering programmatic registration and robust APIs that perfectly complement asynchronous programming for scalable, AI-driven identity solutions.

The Need for Speed: Why Asynchronous Programming Matters for Identity

In today's fast-paced digital world, applications demand instant responses. Whether you're onboarding new users, processing transactions, or verifying identities, delays can lead to lost customers and revenue. Identity verification, in particular, often involves multiple API calls—to check documents, perform liveness detection, or screen against watchlists. Synchronous API calls, where your program waits for each request to complete before moving to the next, can quickly become a bottleneck, especially when dealing with high volumes.

This is where asynchronous programming, specifically Python's asyncio, shines. By allowing your application to initiate multiple tasks concurrently without blocking the main execution thread, asyncio enables significantly higher throughput and responsiveness. For integrating with powerful identity platforms like Didit, which offer robust APIs for ID Verification, Passive & Active Liveness, 1:1 Face Match, and AML Screening, leveraging asyncio is not just an optimization—it's a necessity for building scalable, high-performance systems.

Imagine a scenario where thousands of users are attempting to sign up simultaneously. Each signup requires several identity checks. With synchronous calls, your server would process these one by one, leading to long queues and frustrated users. With asyncio, your application can initiate all these checks almost simultaneously, waiting for the results to come back as they are ready, dramatically improving the user experience and overall system capacity.

Implementing Async API Calls to Didit

Integrating with Didit's API asynchronously involves using an HTTP client that supports asyncio, such as httpx or aiohttp. These libraries allow you to send requests without blocking, making them ideal for high-throughput scenarios.

Let's consider a common use case: initiating multiple identity verification sessions. Didit's API is designed for efficiency, and with asyncio, you can maximize this. For instance, creating verification sessions via POST /v3/session/ can be done concurrently. Each session might involve Didit's ID Verification (OCR, MRZ, barcodes) or Passive & Active Liveness checks.

import asyncio
import httpx

async def create_didit_session(api_key: str, user_id: str):
    url = "https://apx.didit.me/v3/sessions/"
    headers = {"x-api-key": api_key, "Content-Type": "application/json"}
    payload = {"external_id": user_id, "flow_id": "your_flow_id"}
    
    async with httpx.AsyncClient() as client:
        response = await client.post(url, headers=headers, json=payload)
        response.raise_for_status()  # Raise an exception for bad status codes
        return response.json()

async def main():
    didit_api_key = "YOUR_DIDIT_API_KEY"
    user_ids = [f"user_{i}" for i in range(100)] # Simulate 100 users

    tasks = [create_didit_session(didit_api_key, user_id) for user_id in user_ids]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Error creating session for user_{user_ids[i]}: {result}")
        else:
            print(f"Session created for user_{user_ids[i]}: {result['session_id']}")

if __name__ == "__main__":
    asyncio.run(main())

This example demonstrates how to create 100 verification sessions concurrently. Each call to create_didit_session is an awaitable coroutine. asyncio.gather runs them all in parallel, significantly reducing the total time compared to sequential execution.

Handling Didit Webhooks Asynchronously for Real-time Updates

Webhooks are crucial for real-time updates from Didit, informing your system about the completion of ID Verification, the result of an AML Screening, or the outcome of a Proof of Address check. When Didit sends a webhook notification, your application needs to process it quickly without blocking subsequent incoming webhooks. An asynchronous webhook handler is essential for this.

A typical asynchronous Python web framework like FastAPI or Aiohttp is perfect for building an async webhook receiver. When Didit sends a webhook (e.g., a session completion notification), your handler can parse the payload, verify the signature using the secret_shared_key obtained from Didit's webhook configuration, and then enqueue the processing of the event to a background task or message queue without delaying the HTTP response back to Didit.

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib
import asyncio

app = FastAPI()
WEBHOOK_SECRET = "whsec_a1b2c3d4e5f6g7h8i9j0..." # Get this from Didit webhook config

async def process_didit_event(event_data: dict):
    # Simulate asynchronous processing, e.g., updating database, notifying users
    print(f"Processing Didit event: {event_data['event_type']} for session {event_data['session_id']}")
    await asyncio.sleep(2) # Simulate I/O bound task
    print(f"Finished processing event for session {event_data['session_id']}")

@app.post("/didit-webhook")
async def didit_webhook(request: Request):
    signature = request.headers.get("X-Didit-Signature")
    if not signature:
        raise HTTPException(status_code=400, detail="Missing X-Didit-Signature header")

    body = await request.body()
    
    # Verify webhook signature (important for security)
    expected_signature = hmac.new(WEBHOOK_SECRET.encode('utf-8'), body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(signature, expected_signature):
        raise HTTPException(status_code=403, detail="Invalid webhook signature")

    event_data = await request.json()
    
    # Process event asynchronously in the background
    asyncio.create_task(process_didit_event(event_data))
    
    return {"status": "received", "message": "Event accepted for processing"}

This FastAPI example shows how to quickly acknowledge the webhook and then offload the actual processing using asyncio.create_task. This ensures that your webhook endpoint remains highly available and responsive, even if processing individual events takes some time. Didit's webhook configuration allows you to specify a webhook_url and webhook_version, making it easy to set up secure and reliable event delivery.

Navigating Rate Limits with Async Strategies

API rate limits are a critical consideration for any high-throughput integration. Didit, like any robust API provider, enforces rate limiting to maintain stability. Understanding and gracefully handling these limits is paramount to avoid 429 Too Many Requests errors and ensure continuous operation.

Didit provides clear rate limit headers (X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset, Retry-After) in its responses. With asyncio, you can implement sophisticated rate-limiting strategies, such as exponential backoff with jitter and token bucket algorithms, without blocking your entire application.

import asyncio
import httpx
import time

async def make_rate_limited_request(client: httpx.AsyncClient, url: str, headers: dict, payload: dict, retries: int = 5):
    for attempt in range(retries):
        response = await client.post(url, headers=headers, json=payload)
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 2 ** attempt)) # Exponential backoff
            print(f"Rate limit hit. Retrying in {retry_after} seconds...")
            await asyncio.sleep(retry_after)
        else:
            response.raise_for_status()
            return response.json()
    raise Exception("Max retries exceeded for rate-limited request.")

async def main_rate_limited():
    didit_api_key = "YOUR_DIDIT_API_KEY"
    url = "https://apx.didit.me/v3/sessions/"
    headers = {"x-api-key": didit_api_key, "Content-Type": "application/json"}
    
    async with httpx.AsyncClient() as client:
        tasks = []
        for i in range(150): # Simulate exceeding a 300 rpm limit with 150 tasks
            payload = {"external_id": f"user_{i}", "flow_id": "your_flow_id"}
            tasks.append(make_rate_limited_request(client, url, headers, payload))
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"Request {i} failed: {result}")
            else:
                print(f"Request {i} succeeded: {result.get('session_id')}")

if __name__ == "__main__":
    asyncio.run(main_rate_limited())

This example shows an asynchronous exponential backoff strategy. When a 429 is received, the coroutine waits for the specified Retry-After duration (or an exponentially increasing time) before retrying, all without blocking other concurrent tasks. Didit's global limits (300 requests per minute per application for GET/Write/Delete endpoints) and endpoint-specific limits (e.g., 600 rpm for POST /v2/session/) are clearly documented, allowing you to design your async clients accordingly.

How Didit Helps Build High-Throughput Identity Solutions

Didit's architecture is inherently designed to support high-throughput operations, making it an ideal partner for asynchronous integrations. Our platform offers several key advantages:

• Developer-First & AI-Native: Didit provides clean APIs and comprehensive documentation, making it easy to integrate with Python's asyncio. Our AI-native approach means that complex identity verification tasks, from ID Verification to Passive & Active Liveness, are processed efficiently on our end, allowing your application to simply await the results.
• Modular Identity Primitives: Didit's modular design allows you to pick and choose the exact identity checks you need. Whether it's ID Verification (OCR, MRZ), 1:1 Face Match, or Proof of Address, each primitive is accessible via a high-performance API, perfectly suited for concurrent calls.
• Programmatic Registration: Unlike many platforms, Didit allows for programmatic registration, even for AI agents. You can get API credentials in just two API calls, fully headless, without ever opening a browser. This minimizes setup friction and enables automated, high-scale deployment of your identity infrastructure.
• Orchestrated Workflows: Didit's no-code engine allows you to orchestrate complex KYC workflows, reducing the number of individual API calls your application needs to manage while still benefiting from asynchronous processing of the overall session outcome.
• Free Core KYC & Flexible Pricing: Didit offers Free Core KYC, allowing you to get started without upfront costs. Our pay-per-successful-check model and no setup fees mean you only pay for what you use, aligning perfectly with scalable, usage-based architectures enabled by asynchronous processing.

By combining the power of Python asyncio with Didit's robust and developer-friendly API, you can build identity verification systems that are not only secure and compliant but also incredibly fast and scalable, capable of handling the demands of modern digital businesses.

Ready to Get Started?

Ready to see Didit in action? Get a free demo today.

Start verifying identities for free with Didit's free tier.