Metadata-Version: 2.4
Name: primeguardia
Version: 1.0.0
Summary: Official PrimeGuardia Sanctions Screening SDK for Python
Author-email: PrimeGuardia <support@primeguardia.com>
License-Expression: MIT
Project-URL: Homepage, https://primeguardia.com
Project-URL: Documentation, https://docs.primeguardia.com
Project-URL: Bug Tracker, https://github.com/primeguardia/sanctions-sdk-python/issues
Keywords: sanctions,compliance,screening,aml,kyc,ofac,pep,primeguardia
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Typing :: Typed
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.24.0
Requires-Dist: typing-extensions>=4.0.0; python_version < "3.11"
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0.0; extra == "dev"
Requires-Dist: respx>=0.20.0; extra == "dev"
Requires-Dist: black>=23.0.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"
Requires-Dist: ruff>=0.0.260; extra == "dev"
Dynamic: license-file

# 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.

[![PyPI version](https://img.shields.io/pypi/v/primeguardia.svg)](https://pypi.org/project/primeguardia/)
[![Python versions](https://img.shields.io/pypi/pyversions/primeguardia.svg)](https://pypi.org/project/primeguardia/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

## 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

```bash
pip install primeguardia
```

Or with Poetry:

```bash
poetry add primeguardia
```

## Quick Start (5 minutes)

### 1. Get your API key

Sign up at [primeguardia.com](https://primeguardia.com) and get your API key.

### 2. Initialize the client

```python
from primeguardia import PrimeGuardia

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

### 3. Screen an entity

```python
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

```python
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

```python
# 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

```python
# 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

```python
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

```python
# 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

```python
# 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

```python
# 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:

```python
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

```python
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:

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
# ✅ 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

```python
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

```python
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

```python
# ✅ 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

```bash
pytest
```

### Type Checking

```bash
mypy src/primeguardia
```

### Code Formatting

```bash
black src/
ruff check src/
```

## Troubleshooting

### Timeout Issues

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

### Debug Mode

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

### Test Connection

```python
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](CONTRIBUTING.md) for guidelines.
