Python SDK
The official Python SDK for ComplianceLayer provides synchronous and asynchronous clients for the ComplianceLayer API. Features include automatic retries with exponential backoff, built-in rate limiting, comprehensive type hints, and support for all API endpoints.
Overview
The compliancelayer Python package provides two client classes:
ComplianceLayer— synchronous client for scripts, notebooks, and simple integrationsAsyncComplianceLayer— async client for high-throughput applications usingasyncio
Both clients share the same interface and configuration options. The SDK handles authentication, request serialization, response parsing, error mapping, and retry logic automatically.
- Python 3.8+ — supports Python 3.8 through 3.13
- Type hints — full type annotations for IDE autocomplete and static analysis
- Auto-retry — exponential backoff on transient failures (429, 500, 502, 503, 504)
- Rate limiting — automatic request throttling to stay within API limits
- Zero required dependencies — uses
httpxunder the hood (installed automatically)
Installation
Install the SDK from PyPI using pip:
pip install compliancelayerQuick Start
Run your first scan in just a few lines:
from compliancelayer import ComplianceLayer
# Initialize the client with your API key
client = ComplianceLayer(api_key="cl_your_api_key_here")
# Run a scan and wait for results
report = client.scan("example.com")
# Access the results
print(f"{report.domain}: {report.grade} ({report.score}/100)")
print(f"Risk level: {report.risk_level}")
print(f"Issues found: {report.issues_count}")
# Iterate over findings
for issue in report.issues:
print(f" [{issue.severity}] {issue.finding}")
if issue.remediation:
print(f" Fix: {issue.remediation}")The scan() method submits the domain, polls for completion, and returns the full report. By default it polls every 2 seconds and times out after 120 seconds.
Async Client
For applications using asyncio, use the AsyncComplianceLayer client. It supports the same methods but with await syntax:
import asyncio
from compliancelayer import AsyncComplianceLayer
async def main():
# Use as async context manager for automatic cleanup
async with AsyncComplianceLayer(api_key="cl_your_api_key") as client:
# Run a scan
report = await client.scan("example.com")
print(f"{report.domain}: {report.grade} ({report.score}/100)")
# Run multiple scans concurrently
domains = ["example.com", "acme.org", "globex.net"]
tasks = [client.scan(domain) for domain in domains]
reports = await asyncio.gather(*tasks)
for report in reports:
print(f"{report.domain}: {report.grade}")
asyncio.run(main())async with to ensure the HTTP connection pool is properly closed. Without it, you may see resource leak warnings.Scanning
The SDK provides several methods for running security scans. Each method corresponds to a different scanning workflow.
client.scan(domain)Submit a domain for scanning and block until the scan completes. Returns the full ScanReport object. This is the simplest way to run a scan. Internally, it calls scan_async() and polls for completion.
# Simple blocking scan
report = client.scan("example.com")
print(f"Score: {report.score}, Grade: {report.grade}")
# With custom timeout (seconds)
report = client.scan("example.com", timeout=180)
# Access module-level results
for name, module in report.modules.items():
print(f" {name}: {module.score}/100 ({module.grade})")client.scan_async(domain)Submit a domain for scanning and return immediately with a ScanJob object. Use this when you want to manage polling yourself or track multiple scans concurrently.
# Submit scan without waiting
job = client.scan_async("example.com")
print(f"Job ID: {job.job_id}")
print(f"Status: {job.status}") # "queued" or "running"
# Check status manually
job.refresh()
print(f"Status: {job.status}")
# Wait for completion with custom interval
job.wait(poll_interval=3, timeout=120)
# Get the report once complete
if job.status == "completed":
report = job.get_report()
print(f"Score: {report.score}")
elif job.status == "failed":
print(f"Scan failed: {job.failure_reason}")client.free_scan(domain)Run a free scan that does not require authentication or count against your quota. Free scans have reduced module coverage and are rate limited to 5 per hour per IP. No API key is required.
from compliancelayer import ComplianceLayer
# Free scan - no API key needed
client = ComplianceLayer()
report = client.free_scan("example.com")
print(f"{report.domain}: {report.grade} ({report.score}/100)")
print(f"Modules scanned: {len(report.modules)}")ScanJob Methods
The ScanJob object returned by scan_async() provides methods for tracking scan progress:
| Method | Returns | Description |
|---|---|---|
job.refresh() | ScanJob | Fetch the latest status from the API and update the job object in place |
job.wait() | ScanJob | Block until the scan completes or times out. Accepts poll_interval and timeout kwargs |
job.get_report() | ScanReport | Fetch the full report for a completed scan. Raises ScanNotComplete if still running |
job.is_complete | bool | Property that returns True if the scan has finished (completed or failed) |
Batch Operations
Scan multiple domains in a single API call. Batch operations are more efficient than individual scans and support up to 50 domains per request.
# Scan multiple domains at once (up to 50)
domains = [
"example.com",
"acme.org",
"globex.net",
"initech.io",
"umbrella.co"
]
# Submit batch scan
batch = client.batch_scan(domains)
print(f"Batch ID: {batch.batch_id}")
print(f"Total domains: {batch.total}")
# Wait for all scans to complete
batch.wait()
# Access individual results
for result in batch.results:
print(f"{result.domain}: {result.grade} ({result.score}/100)")
# Get summary statistics
print(f"Average score: {batch.average_score}")
print(f"Completed: {batch.completed_count}/{batch.total}")
print(f"Failed: {batch.failed_count}")# Compare security posture across domains
comparison = client.compare([
"example.com",
"competitor1.com",
"competitor2.com"
])
# Overall comparison
print(f"Best: {comparison.best.domain} ({comparison.best.score})")
print(f"Worst: {comparison.worst.domain} ({comparison.worst.score})")
print(f"Average: {comparison.average_score}")
# Module-level comparison
for module_name, scores in comparison.by_module.items():
print(f"\n{module_name}:")
for domain, score in scores.items():
print(f" {domain}: {score}/100")Webhooks
Manage webhook endpoints programmatically. Create, list, test, and inspect webhook deliveries.
# Create a new webhook endpoint
webhook = client.webhooks.create(
url="https://api.yourapp.com/webhooks/compliancelayer",
events=["scan.completed", "alert.triggered"],
description="Production webhook for scan results"
)
print(f"Webhook ID: {webhook.id}")
print(f"Secret: {webhook.secret}") # Shown only once!
print(f"Status: {webhook.status}") # "active"Domain Monitoring
Add domains to your monitoring list for scheduled scans and automatic alerts when security posture changes.
# Add a domain to monitoring
domain = client.domains.add(
domain="example.com",
scan_interval="weekly", # "daily", "weekly", "monthly"
alert_threshold=70 # Alert if score drops below this
)
print(f"Domain ID: {domain.id}")
print(f"Monitoring: {domain.domain}")
print(f"Scan interval: {domain.scan_interval}")
print(f"Next scan: {domain.next_scan_at}")Badges
Generate embeddable security badges for your domains. Badges show the current security grade and update automatically with each scan.
# Get the badge URL for a domain
url = client.badge_url("example.com")
print(f"Badge URL: {url}")
# https://api.compliancelayer.net/v1/badge/example.com
# Use in Markdown
markdown = f""
print(markdown)
# Get the raw SVG content
svg = client.badge_svg("example.com")
with open("badge.svg", "w") as f:
f.write(svg)
# Customize badge style
url = client.badge_url(
"example.com",
style="flat", # "flat", "flat-square", "plastic"
label="Security" # Custom left-side label
)Configuration
Both ComplianceLayer and AsyncComplianceLayer accept the same constructor options:
| Parameter | Type | Default | Description |
|---|---|---|---|
api_key | str | None | Your ComplianceLayer API key (starts with cl_). If not provided, reads from COMPLIANCELAYER_API_KEY environment variable. |
base_url | str | "https://api.compliancelayer.net/v1" | Base URL for the API. Override for self-hosted instances or local development. |
timeout | float | 30.0 | HTTP request timeout in seconds for individual API calls. |
poll_interval | float | 2.0 | Seconds between status polls when using blocking scan() or job.wait(). |
poll_timeout | float | 120.0 | Maximum seconds to wait for a scan to complete when polling. |
max_retries | int | 3 | Number of automatic retries on transient errors (429, 5xx). Uses exponential backoff. |
from compliancelayer import ComplianceLayer
# Fully customized client
client = ComplianceLayer(
api_key="cl_your_api_key",
base_url="https://api.compliancelayer.net/v1",
timeout=60.0, # 60s HTTP timeout
poll_interval=3.0, # Poll every 3 seconds
poll_timeout=180.0, # Wait up to 3 minutes for scans
max_retries=5 # Retry up to 5 times on failures
)
# Or use environment variable for API key
# export COMPLIANCELAYER_API_KEY=cl_your_api_key
client = ComplianceLayer() # Reads from envapi_key is not passed to the constructor, the SDK automatically reads from the COMPLIANCELAYER_API_KEY environment variable. This is the recommended approach for production deployments.Error Handling
The SDK raises typed exceptions for different error conditions. All exceptions inherit from ComplianceLayerError, so you can catch all SDK errors with a single except clause or handle specific errors individually.
from compliancelayer import ComplianceLayer
from compliancelayer.exceptions import (
ComplianceLayerError, # Base exception for all SDK errors
APIError, # Generic API error (non-2xx response)
AuthenticationError, # Invalid or missing API key (401)
ForbiddenError, # Insufficient permissions (403)
NotFoundError, # Resource not found (404)
RateLimitError, # Rate limit exceeded (429)
QuotaExceededError, # Scan quota exhausted
ValidationError, # Invalid request parameters (422)
ScanTimeoutError, # Scan polling timed out
ScanFailedError, # Scan execution failed
ServerError, # Server-side error (5xx)
ConnectionError, # Network connectivity issue
)
client = ComplianceLayer(api_key="cl_your_api_key")
try:
report = client.scan("example.com")
print(f"Score: {report.score}")
except AuthenticationError:
print("Invalid API key. Check your credentials.")
except QuotaExceededError as e:
print(f"Scan quota exhausted. Resets at: {e.reset_at}")
print(f"Current usage: {e.used}/{e.limit}")
except RateLimitError as e:
print(f"Rate limited. Retry after {e.retry_after} seconds.")
except ScanTimeoutError:
print("Scan took too long. Try again or increase poll_timeout.")
except ScanFailedError as e:
print(f"Scan failed: {e.reason}")
print(f"Domain: {e.domain}")
except ValidationError as e:
print(f"Invalid input: {e.detail}")
for error in e.errors:
print(f" Field '{error['field']}': {error['message']}")
except ServerError:
print("ComplianceLayer server error. Try again later.")
except ConnectionError:
print("Network error. Check your internet connection.")
except ComplianceLayerError as e:
# Catch-all for any SDK error
print(f"Unexpected error: {e}")Exception Hierarchy
| Exception | HTTP Status | Description |
|---|---|---|
ComplianceLayerError | - | Base class for all SDK exceptions |
APIError | Any non-2xx | Generic API error with status code and message |
AuthenticationError | 401 | Invalid, missing, or expired API key |
ForbiddenError | 403 | Valid key but insufficient plan permissions |
NotFoundError | 404 | Requested resource does not exist |
ValidationError | 422 | Request failed validation (invalid domain, missing fields) |
RateLimitError | 429 | Too many requests; includes retry_after seconds |
QuotaExceededError | 429 | Monthly scan quota exhausted; includes reset date |
ServerError | 500-599 | Server-side error; safe to retry |
ScanTimeoutError | - | Polling timed out before scan completed |
ScanFailedError | - | Scan execution failed (invalid domain, DNS error, etc.) |
ConnectionError | - | Network error (DNS resolution, timeout, refused) |
Advanced Usage
Custom HTTP Client
Pass a custom httpx client for advanced configuration like custom TLS certificates, proxies, or connection pooling:
import httpx
from compliancelayer import ComplianceLayer
# Custom HTTP client with proxy and TLS settings
http_client = httpx.Client(
proxy="http://proxy.corp.example.com:8080",
verify="/path/to/custom/ca-bundle.crt",
limits=httpx.Limits(
max_connections=20,
max_keepalive_connections=10
)
)
client = ComplianceLayer(
api_key="cl_your_api_key",
http_client=http_client
)Logging
The SDK uses Python's standard logging module. Enable debug logging to see all HTTP requests and responses:
import logging
# Enable debug logging for the SDK
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("compliancelayer")
logger.setLevel(logging.DEBUG)
# Now all API requests will be logged:
# DEBUG:compliancelayer:POST /v1/scan {"domain": "example.com"}
# DEBUG:compliancelayer:200 OK (234ms)
# DEBUG:compliancelayer:GET /v1/scan/jobs/12345
# DEBUG:compliancelayer:200 OK {"status": "completed", ...}Pagination
List endpoints that return many results use cursor-based pagination. The SDK provides an iterator that handles pagination automatically:
# Automatic pagination - iterates through all pages
for domain in client.domains.list():
print(f"{domain.domain}: {domain.latest_score}")
# Manual pagination with page size control
page = client.domains.list(limit=25)
print(f"Total domains: {page.total}")
print(f"Page 1 of {page.total_pages}")
for domain in page.items:
print(f" {domain.domain}")
# Get next page
if page.has_next:
next_page = page.next()Webhook Signature Verification
The SDK includes a utility for verifying incoming webhook signatures in your server:
from compliancelayer.webhooks import verify_signature
# In your Flask/FastAPI/Django webhook handler
def handle_webhook(request):
payload = request.get_data(as_text=True)
signature = request.headers.get("X-ComplianceLayer-Signature")
secret = "whsec_your_webhook_secret"
# Verify the signature (raises InvalidSignatureError if invalid)
verify_signature(payload, signature, secret)
# Safe to process the event
event = json.loads(payload)
print(f"Event: {event['type']}")