kawaa 0.1.0

Official Python SDK for Kawaa Email Verification API

Kawaa Python SDK

Official Python SDK for Kawaa Email Verification API.

Features

• Verify single emails or batch lists
• Automatic retry with exponential backoff
• Both sync and async clients
• Full type hints for IDE support
• Webhook signature verification
• Framework integrations (Flask, FastAPI)

Installation

pip install kawaa

Quick Start

from kawaa import Kawaa

# Create client with your API key
client = Kawaa("your-api-key")

# Verify a single email
result = client.verify("[e-mail]")
print(f"Status: {result.status}")  # valid, invalid, risky, or unknown
print(f"Score: {result.score}")     # 0-100 quality score

if result.is_valid:
    print("Email is valid!")

if result.flags.disposable:
    print("Warning: disposable email")

Single Email Verification

from kawaa import Kawaa

client = Kawaa("your-api-key")

# Basic verification
result = client.verify("[e-mail]")

# Access results
print(f"Email: {result.email}")
print(f"Status: {result.status}")      # VerificationStatus enum
print(f"Score: {result.score}")        # 0-100
print(f"Disposable: {result.flags.disposable}")
print(f"Role account: {result.flags.role}")
print(f"Free provider: {result.flags.free_provider}")
print(f"Catch-all: {result.flags.catch_all}")

# AI-suggested correction (if typo detected)
if result.suggestion:
    print(f"Did you mean: {result.suggestion}")

# Customize verification options
result = client.verify(
    "[e-mail]",
    check_mx=True,        # Check MX records
    check_smtp=True,      # Perform SMTP verification
    check_catch_all=True, # Detect catch-all domains
    use_ai=True,          # Use AI for typo detection
)

Batch Verification

For verifying many emails at once:

from kawaa import Kawaa

client = Kawaa("your-api-key")

emails = [
    "[e-mail]",
    "[e-mail]",
    "invalid@",
    "[e-mail]",  # duplicate - will be removed
]

# Start batch job
job = client.verify_batch(emails)
print(f"Job ID: {job.job_id}")
print(f"Duplicates removed: {job.duplicates_removed}")
print(f"Credits used: {job.credits_used}")

# Poll for completion
result = client.wait_for_job(job.job_id, timeout=300)
print(f"Valid: {result.valid_count}")
print(f"Invalid: {result.invalid_count}")
print(f"Risky: {result.risky_count}")

# Access individual results
for r in result.results:
    print(f"{r.email}: {r.status} (score: {r.score})")

One-liner with wait

# Start job and wait for completion in one call
result = client.verify_batch_and_wait(
    emails,
    poll_interval=2.0,  # Check every 2 seconds
    timeout=600,        # Wait up to 10 minutes
)

Using webhooks

# Get notified when job completes
job = client.verify_batch(
    emails,
    webhook_url="https://your-server.com/webhook",
    webhook_secret="your-secret",  # Optional, for signature verification
)

Download Results

# Download as JSON
results = client.download_results(job_id)
for r in results.results:
    print(f"{r.email}: {r.status}")

# Download as CSV
csv_data = client.download_results(job_id, format="csv")
print(csv_data)  # CSV string

# For large result sets, you'll get a presigned URL
if results.download_url:
    print(f"Download from: {results.download_url}")

Async Client

For async/await support:

import asyncio
from kawaa import AsyncKawaa

async def main():
    async with AsyncKawaa("your-api-key") as client:
        # Single verification
        result = await client.verify("[e-mail]")
        print(f"Status: {result.status}")

        # Batch verification
        job = await client.verify_batch(emails)
        result = await client.wait_for_job(job.job_id)
        print(f"Valid: {result.valid_count}")

asyncio.run(main())

Webhook Handling

Verify webhook signatures and parse events:

from kawaa.webhooks import parse_webhook, verify_signature

# Verify signature
is_valid = verify_signature(
    payload=request.body,
    signature=request.headers["X-Kawaa-Signature"],
    secret="your-webhook-secret",
)

# Parse webhook event
event = parse_webhook(
    payload=request.body,
    signature=request.headers.get("X-Kawaa-Signature"),
    secret="your-webhook-secret",  # Optional
)

print(f"Job {event.job_id} completed!")
print(f"Valid: {event.valid}, Invalid: {event.invalid}")
print(f"Download: {event.download_url}")

Flask Integration

from flask import Flask
from kawaa.webhooks import flask_webhook_handler

app = Flask(__name__)

@app.route("/webhook", methods=["POST"])
@flask_webhook_handler("your-webhook-secret")
def handle_webhook(event):
    print(f"Job {event.job_id} completed!")
    return "", 200

FastAPI Integration

from fastapi import FastAPI, Depends
from kawaa.webhooks import fastapi_webhook_handler, WebhookEvent

app = FastAPI()
get_event = fastapi_webhook_handler("your-webhook-secret")

@app.post("/webhook")
async def handle_webhook(event: WebhookEvent = Depends(get_event)):
    print(f"Job {event.job_id} completed!")
    return {"status": "ok"}

Account Information

account = client.get_account()
print(f"Credits remaining: {account.credits_remaining}")
print(f"Plan: {account.plan}")

Error Handling

from kawaa import Kawaa
from kawaa import (
    KawaaError,
    AuthenticationError,
    RateLimitError,
    InsufficientCreditsError,
    ValidationError,
    NotFoundError,
    TimeoutError,
)

client = Kawaa("your-api-key")

try:
    result = client.verify("[e-mail]")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except InsufficientCreditsError:
    print("Not enough credits")
except ValidationError as e:
    print(f"Invalid request: {e.message}")
except TimeoutError:
    print("Request timed out")
except KawaaError as e:
    print(f"API error: {e.message}")

Configuration

client = Kawaa(
    api_key="your-api-key",
    base_url="https://api.kawaa.com",  # Custom API URL
    timeout=30.0,                       # Request timeout in seconds
    max_retries=3,                      # Max retry attempts
)

Context Manager

The client can be used as a context manager for automatic cleanup:

with Kawaa("your-api-key") as client:
    result = client.verify("[e-mail]")
# Connection closed automatically

Models Reference

VerificationResult

Field	Type	Description
email	str	The verified email
status	VerificationStatus	valid, invalid, risky, unknown
score	int	Quality score 0-100
flags	VerificationFlags	Email characteristics
suggestion	str	AI-suggested correction
verification	VerificationDetails	Technical details

VerificationFlags

Field	Type	Description
disposable	bool	Temporary email service
role	bool	Role account (info@, support@)
free_provider	bool	Free email provider
catch_all	bool	Domain accepts all emails

JobStatusResult

Field	Type	Description
job_id	str	Unique job identifier
status	JobStatus	pending, processing, completed, failed
total_emails	int	Total emails in job
processed_emails	int	Emails processed so far
valid_count	int	Valid email count
invalid_count	int	Invalid email count
risky_count	int	Risky email count
progress_percent	float	Completion percentage
results	List[VerificationResult]	Individual results

License

MIT License - see LICENSE for details.

Support

• Documentation: https://docs.kawaa.com
• Email: [e-mail]
• Issues: https://github.com/kawaa/kawaa-python/issues