PrimeGuardia Python SDK

Official Python SDK for PrimeGuardia's Sanctions Screening API. Screen individuals and entities against global sanctions lists, PEPs, and watchlists with type-safe, Pythonic interfaces.

Features

• ✅ Full type hints for excellent IDE support
• ⚡ Both sync and async clients
• 🔄 Automatic retry with exponential backoff
• 🛡️ Typed exceptions for precise error handling
• 📊 Dataclasses for structured responses
• 🎯 Context managers for resource management
• 🔍 Bulk screening (up to 1,000 entities)
• 📡 Real-time monitoring support
• 🚀 Production-ready with connection pooling
• 🐍 Pythonic API design

Installation

pip install primeguardia

Or with Poetry:

poetry add primeguardia

Quick Start (5 minutes)

1. Get your API key

Sign up at primeguardia.com and get your API key.

2. Initialize the client

from primeguardia import PrimeGuardia

client = PrimeGuardia(api_key="your-api-key-here")

3. Screen an entity

result = client.screen(name="John Doe", email="john@example.com")

if result.should_block:
    print("⚠️  SHOULD BE BLOCKED — review required")
    print(f"Risk level: {result.risk_assessment}")
    print(f"Matches: {result.matches}")
else:
    print("✅ Clear - no matches found")

That's it! You're screening entities in 3 simple steps.

Usage Examples

Basic Screening

from primeguardia import PrimeGuardia

client = PrimeGuardia(api_key="your-api-key")

# Screen by name
result = client.screen(name="Vladimir Putin")
print(f"Match: {result.match}")
print(f"Confidence: {result.confidence}")
print(f"Risk: {result.risk_assessment}")

# Use convenience properties
if result.is_high_risk:
    print("⚠️  HIGH RISK DETECTED!")

if result.is_clear:
    print("✅ All clear")

# Screen with additional context
result = client.screen(
    name="John Smith",
    email="john@example.com",
    country="US",
    date_of_birth="1980-01-01",
    metadata={"customer_id": "CUST-12345"}
)

Context Manager

# Automatically closes connection when done
with PrimeGuardia(api_key="your-key") as client:
    result = client.screen(name="John Doe")
    print(result.risk_assessment)
# Connection closed automatically

Bulk Screening

# Screen up to 1,000 entities in one request
results = client.bulk_screen(
    names=["John Doe", "Jane Smith", "Vladimir Putin"],
    emails=["john@example.com", "jane@example.com", "president@kremlin.ru"]
)

print(f"Processed {results.processed} entities")
print(f"Time: {results.processing_time_ms}ms")
print(f"High-risk matches: {results.high_risk_count}")
print(f"Total matches: {results.matches_count}")

# Iterate through results
for result in results.results:
    if result.match:
        print(f"⚠️  {result.name}: {result.risk_level} risk (score: {result.score})")

Async Client

import asyncio
from primeguardia import AsyncPrimeGuardia

async def main():
    async with AsyncPrimeGuardia(api_key="your-key") as client:
        # All methods support await
        result = await client.screen(name="John Doe")

        if result.should_block:
            print(f"Blocked! Risk: {result.risk_assessment}")

        # Concurrent requests
        tasks = [
            client.screen(name="Person 1"),
            client.screen(name="Person 2"),
            client.screen(name="Person 3"),
        ]
        results = await asyncio.gather(*tasks)

        for result in results:
            print(f"Match: {result.match}, Score: {result.score}")

# Run async code
asyncio.run(main())

Search Database

# Search sanctions database
results = client.search(
    query="putin",
    limit=20,
    sources=["ofac", "eu_sanctions"]
)

print(f"Found {results.total} matches")

# Check if there are more results
if results.has_more:
    print("More results available. Increase limit or offset.")

# Iterate through entities
for entity in results.results:
    print(f"{entity.name}")
    print(f"  Sources: {', '.join(entity.source_dataset)}")
    print(f"  Countries: {', '.join(entity.countries or [])}")

# Get specific entity
entity = client.get_entity(12345)
print(entity.name, entity.source_dataset)

Continuous Monitoring

# Add entity to monitoring
monitored = client.add_monitoring(
    name="Suspicious Person",
    email="suspicious@example.com",
    frequency=24,  # Check every 24 hours
    metadata={"internal_id": "CUST-12345"}
)

print(f"Now monitoring entity {monitored.id}")

# Get all monitored entities
entities = client.get_monitored_entities()
print(f"Monitoring {len(entities)} entities")

# Note: Full monitoring API coming in next version

Account Management

# Get profile
profile = client.get_profile()
print(f"Client: {profile.client_name}")
print(f"Tier: {profile.tier}")
print(f"Status: {profile.subscription_status}")
print(f"Usage: {profile.usage_percentage:.1f}%")

# Check if subscription is active
if profile.is_active:
    print("✅ Subscription active")

# Check quota
quota = client.get_quota_status()
print(f"Used: {quota.used}/{quota.limit} ({quota.percentage}%)")
print(f"Remaining: {quota.remaining} calls")

# Check quota status with convenience methods
if quota.is_critical:
    print("⚠️  CRITICAL: >95% of quota used!")
elif quota.is_low:
    print("⚠️  Warning: >80% of quota used")

if quota.is_exceeded:
    print("❌ Quota exceeded!")

# Get available datasets
datasets = client.get_datasets()
for dataset in datasets:
    status = "✅" if dataset.available else "❌ Upgrade required"
    print(f"{status} {dataset.name}: {dataset.record_count:,} records")

Error Handling

The SDK provides typed exceptions for precise error handling:

from primeguardia import (
    PrimeGuardia,
    AuthenticationError,
    QuotaExceededError,
    RateLimitError,
    ValidationError,
    PrimeGuardiaError
)

client = PrimeGuardia(api_key="your-key")

try:
    result = client.screen(name="John Doe")
except AuthenticationError as e:
    print(f"Invalid API key: {e}")
    # Update API key
except QuotaExceededError as e:
    print(f"Quota exceeded: {e}")
    # Upgrade plan or wait for reset
except RateLimitError as e:
    print(f"Rate limit exceeded: {e}")
    if e.retry_after:
        print(f"Retry after {e.retry_after} seconds")
        time.sleep(e.retry_after)
except ValidationError as e:
    print(f"Validation error: {e}")
    if e.details:
        print(f"Details: {e.details}")
except PrimeGuardiaError as e:
    print(f"API error ({e.status_code}): {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Configuration

client = PrimeGuardia(
    api_key="your-api-key",          # Required
    base_url="https://api.primeguardia.com",  # Optional
    timeout=30.0,                     # Request timeout in seconds
    max_retries=3,                    # Max retry attempts
    debug=False                       # Enable debug logging
)

Type Safety

Full type hints for excellent IDE support:

from primeguardia import (
    PrimeGuardia,
    ScreeningResult,
    BulkScreeningResult,
    ClientProfile,
    ConfidenceLevel,
    RiskLevel,
)

client: PrimeGuardia = PrimeGuardia(api_key="your-key")

# Type checking works perfectly
result: ScreeningResult = client.screen(name="John Doe")
confidence: ConfidenceLevel = result.confidence  # "high" | "medium" | "low" | "none"
risk: RiskLevel = result.risk_assessment  # "HIGH" | "MEDIUM" | "LOW" | "CLEAR"

# Dataclass properties
profile: ClientProfile = client.get_profile()
usage_pct: float = profile.usage_percentage
is_active: bool = profile.is_active

Framework Integration

Django View

from django.http import JsonResponse
from django.views import View
from primeguardia import PrimeGuardia, PrimeGuardiaError

class ScreeningView(View):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.client = PrimeGuardia(api_key=settings.PRIMEGUARDIA_API_KEY)

    def post(self, request):
        name = request.POST.get('name')
        email = request.POST.get('email')

        try:
            result = self.client.screen(name=name, email=email)

            if result.should_block:
                return JsonResponse({
                    'allowed': False,
                    'reason': 'Sanctions screening failed',
                    'risk_level': result.risk_assessment
                }, status=403)

            return JsonResponse({'allowed': True})

        except PrimeGuardiaError as e:
            return JsonResponse({'error': str(e)}, status=500)

Flask API

from flask import Flask, request, jsonify
from primeguardia import PrimeGuardia
import os

app = Flask(__name__)
client = PrimeGuardia(api_key=os.getenv('PRIMEGUARDIA_API_KEY'))

@app.route('/api/screen', methods=['POST'])
def screen():
    data = request.get_json()

    result = client.screen(
        name=data.get('name'),
        email=data.get('email')
    )

    return jsonify({
        'match': result.match,
        'should_block': result.should_block,
        'risk_assessment': result.risk_assessment,
        'confidence': result.confidence,
        'score': result.score
    })

@app.route('/api/health')
def health():
    try:
        client.test_connection()
        return jsonify({'status': 'ok'})
    except Exception as e:
        return jsonify({'status': 'error', 'message': str(e)}), 500

FastAPI

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from primeguardia import AsyncPrimeGuardia, PrimeGuardiaError
import os

app = FastAPI()
client = AsyncPrimeGuardia(api_key=os.getenv('PRIMEGUARDIA_API_KEY'))

class ScreenRequest(BaseModel):
    name: str
    email: str | None = None

@app.post("/api/screen")
async def screen(request: ScreenRequest):
    try:
        result = await client.screen(
            name=request.name,
            email=request.email
        )

        return {
            "match": result.match,
            "should_block": result.should_block,
            "risk_assessment": result.risk_assessment,
            "confidence": result.confidence
        }
    except PrimeGuardiaError as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.on_event("shutdown")
async def shutdown():
    await client.close()

Celery Task

from celery import Celery
from primeguardia import PrimeGuardia
import os

app = Celery('tasks', broker='redis://localhost:6379')
client = PrimeGuardia(api_key=os.getenv('PRIMEGUARDIA_API_KEY'))

@app.task
def screen_user(user_id, name, email):
    """Background task to screen a user"""
    result = client.screen(name=name, email=email)

    if result.should_block:
        # Handle blocked user
        send_alert(user_id, result.risk_assessment)
        block_user_account(user_id)

    return {
        'user_id': user_id,
        'should_block': result.should_block,
        'score': result.score
    }

@app.task
def bulk_screen_users(users):
    """Background task to screen multiple users"""
    names = [u['name'] for u in users]
    emails = [u['email'] for u in users]

    results = client.bulk_screen(names=names, emails=emails)

    # Process results
    for user, result in zip(users, results.results):
        if result.match:
            handle_match(user['id'], result)

Best Practices

1. Use Environment Variables

import os
from primeguardia import PrimeGuardia

# ✅ Good
client = PrimeGuardia(api_key=os.getenv('PRIMEGUARDIA_API_KEY'))

# ❌ Bad - never hardcode
client = PrimeGuardia(api_key='abc123...')

2. Use Context Managers

# ✅ Good - automatically closes connection
with PrimeGuardia(api_key=api_key) as client:
    result = client.screen(name="John Doe")

# ❌ Less optimal - manual cleanup
client = PrimeGuardia(api_key=api_key)
result = client.screen(name="John Doe")
client.close()  # Easy to forget!

3. Cache Results

from functools import lru_cache
from primeguardia import PrimeGuardia

client = PrimeGuardia(api_key="your-key")

@lru_cache(maxsize=1000)
def screen_cached(name: str, email: str):
    """Cache screening results for 1000 unique entities"""
    result = client.screen(name=name, email=email)
    return result.should_block, result.risk_assessment

# Or use Redis/Memcached for distributed caching

4. Handle Errors Gracefully

def safe_screen(name, email):
    """Fail-safe screening with fallback"""
    try:
        result = client.screen(name=name, email=email)
        return result.should_block
    except QuotaExceededError:
        logger.error("Quota exceeded!")
        # Fail safely - don't block legitimate users
        return False
    except PrimeGuardiaError as e:
        logger.error(f"Screening failed: {e}")
        # Decide on failure mode based on compliance requirements
        return False  # or True for fail-closed

5. Use Bulk Operations

# ✅ Good - bulk operation
users = [{"name": "User 1", "email": "user1@example.com"}, ...]
results = client.bulk_screen(
    names=[u["name"] for u in users],
    emails=[u["email"] for u in users]
)

# ❌ Less efficient - individual calls
for user in users:
    result = client.screen(name=user["name"], email=user["email"])

Development

Running Tests

pytest

Type Checking

mypy src/primeguardia

Code Formatting

black src/
ruff check src/

Troubleshooting

Timeout Issues

# Increase timeout for slow connections
client = PrimeGuardia(api_key="your-key", timeout=60.0)

Debug Mode

# Enable debug logging
client = PrimeGuardia(api_key="your-key", debug=True)

Test Connection

try:
    client.test_connection()
    print("✅ Connection successful")
except AuthenticationError:
    print("❌ Invalid API key")
except Exception as e:
    print(f"❌ Connection failed: {e}")

Requirements

• Python 3.8+
• httpx >= 0.24.0

Support

• 📧 Email: support@primeguardia.com
• 📚 Documentation: https://docs.primeguardia.com
• 🐛 Issues: https://github.com/primeguardia/sanctions-sdk-python/issues

License

MIT © PrimeGuardia

Contributing

Contributions welcome! Please see CONTRIBUTING.md for guidelines.