vettly 0.1.9
Content moderation SDK for apps. Filtering, reporting, blocking, and audit trails.
Requires Python: >=3.8
Keywords: audit-trail, content-filtering, content-moderation, content-moderation-api, content-safety, harmful-content, moderation, nsfw-detection, platform-compliance, python, sdk, toxicity-detection, trust-safety, ugc-moderation, ugc-safety, user-generated-content

# vettly

Decision infrastructure for Python. Every decision recorded, versioned, and auditable—with full async support.

## The Problem

Your platform makes thousands of automated decisions daily. But when a user appeals—or a regulator asks—can you prove:
- **What policy** was applied?
- **Why** that action was taken?
- **What evidence** supports the decision?

Classification APIs return `True` or `False`. That's not enough for DSA Article 17 compliance. That's not enough when your Trust & Safety team needs to explain decisions months later.

## The Solution

Vettly is **decision infrastructure**. Every API call returns an `id` that links to:
- The exact policy version applied
- Category scores and thresholds that triggered the action
- Content fingerprint for tamper-evident verification

```python
from vettly import ModerationClient

client = ModerationClient("vettly_live_...")

result = client.check(
    content=user_content,
    policy_id="community-guidelines-v2"
)

# Store decision ID with your content
db.posts.create(
    content=user_content,
    moderation_decision_id=result.id,  # Link to audit record
    action=result.action.value
)

# Later: retrieve for compliance review
decision = client.get_decision(result.id)
print(decision.policy.version)  # Exact policy version
```

## Installation

```bash
pip install vettly
```

## Quick Start

```python
from vettly import ModerationClient

client = ModerationClient("vettly_live_...")

result = client.check(
    content="User-generated text",
    policy_id="community-safe"
)

if result.action == "block":
    print(f"Blocked: {result.id}")
else:
    print(f"Allowed: {result.action}")
```

## Get Your API Key

1. Sign up at [vettly.dev](https://vettly.dev)
2. Go to Dashboard > API Keys
3. Create and copy your key

---

## Core Features

### Text Decisions

```python
result = client.check(
    content="User-generated text",
    policy_id="community-safe",
    content_type="text",
    metadata={"user_id": "user_123"},
    user_id="user_123",
    request_id="unique-request-id"  # For idempotency
)

print(result.id)          # Decision ID for audit trail
print(result.action)      # Action.ALLOW | Action.WARN | Action.FLAG | Action.BLOCK
print(result.safe)        # bool
print(result.flagged)     # bool
print(result.categories)  # List[CategoryResult]
print(result.latency_ms)  # Response time
```

### Image Decisions

```python
# From URL
result = client.check_image(
    image_url="https://example.com/image.jpg",
    policy_id="strict"
)

# From base64 data URI
result = client.check_image(
    image_url="data:image/jpeg;base64,/9j/4AAQ...",
    policy_id="strict"
)
```

### Batch Operations

Process multiple items efficiently:

```python
from vettly import BatchItem

# Synchronous batch
batch_result = client.batch_check(
    policy_id="community-safe",
    items=[
        BatchItem(id="post-1", content="First post"),
        BatchItem(id="post-2", content="Second post"),
        BatchItem(id="post-3", content="Third post"),
    ]
)

for result in batch_result.results:
    print(f"{result.id}: {result.action}")

# Async batch with webhook delivery
async_batch = client.batch_check_async(
    policy_id="community-safe",
    items=[...],
    webhook_url="https://your-app.com/webhooks/batch-complete"
)

print(f"Batch ID: {async_batch.batch_id}")
```

### Policy Replay

Re-evaluate historical decisions with different policies:

```python
# User appeals a decision - test with updated policy
original_decision = client.get_decision("original-decision-id")
replay = client.replay_decision(
    decision_id="original-decision-id",
    policy_id="community-guidelines-v3"  # New policy version
)

# Compare outcomes
print(f"Original: {original_decision.action}")
print(f"With new policy: {replay.action}")
```

### Dry Run

Test policies without making provider calls:

```python
dry_run = client.dry_run(
    policy_id="new-policy-id",
    mock_scores={
        "hate_speech": 0.7,
        "harassment": 0.3,
        "violence": 0.1
    }
)

print(dry_run["evaluation"]["action"])  # What action would be taken
```

---

## Async Support

Full async/await support with `AsyncModerationClient`:

```python
from vettly import AsyncModerationClient

async with AsyncModerationClient("vettly_live_...") as client:
    result = await client.check(
        content="User content",
        policy_id="community-safe"
    )

    if result.action == "block":
        print(f"Blocked: {result.id}")
```

### With FastAPI

```python
from fastapi import FastAPI, HTTPException
from vettly import AsyncModerationClient

app = FastAPI()
client = AsyncModerationClient("vettly_live_...")

@app.post("/comments")
async def create_comment(content: str):
    result = await client.check(
        content=content,
        policy_id="community-safe"
    )

    if result.action == "block":
        raise HTTPException(
            status_code=403,
            detail={
                "error": "Content blocked",
                "decision_id": result.id
            }
        )

    # Save with audit trail
    await db.comments.create(
        content=content,
        moderation_decision_id=result.id
    )

    return {"status": "ok", "decision_id": result.id}

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

### With Django (async views)

```python
from django.http import JsonResponse
from vettly import AsyncModerationClient

client = AsyncModerationClient("vettly_live_...")

async def create_post(request):
    content = request.POST.get("content")

    result = await client.check(
        content=content,
        policy_id="community-safe"
    )

    if result.action == "block":
        return JsonResponse(
            {"error": "Content blocked", "decision_id": result.id},
            status=403
        )

    # Save post
    post = await Post.objects.acreate(
        content=content,
        moderation_decision_id=result.id
    )

    return JsonResponse({"id": post.id})
```

---

## Decision Retrieval & Audit

### Get Decision Details

```python
decision = client.get_decision("decision-uuid")

print(decision.id)
print(decision.action)
print(decision.categories)
print(decision.policy.id)
print(decision.policy.version)
print(decision.created_at)
```

### List Recent Decisions

```python
decisions = client.list_decisions(limit=50, offset=0)

for d in decisions:
    print(f"{d.id}: {d.action} ({d.created_at})")
```

### Get cURL for Debugging

```python
curl_command = client.get_curl_command("decision-uuid")
print(curl_command)
# curl -X POST https://api.vettly.dev/v1/check ...
```

---

## Policy Management

### Create or Update Policy

```python
policy = client.create_policy(
    policy_id="my-policy",
    yaml_content="""
name: My Community Policy
version: "1.0"
rules:
  - category: hate_speech
    threshold: 0.7
    provider: openai
    action: block
  - category: harassment
    threshold: 0.8
    provider: openai
    action: flag
fallback:
  provider: mock
  on_timeout: true
  timeout_ms: 5000
"""
)

print(policy.version)  # Immutable version hash
```

### List Policies

```python
policies = client.list_policies()

for p in policies:
    print(f"{p.id}: v{p.version}")
```

---

## Webhooks

### Register a Webhook

```python
webhook = client.register_webhook(
    url="https://your-app.com/webhooks/vettly",
    events=["decision.blocked", "decision.flagged"],
    description="Production webhook for blocked content"
)

print(webhook.id)
print(webhook.secret)  # Use for signature verification
```

### Webhook Signature Verification

```python
from flask import Flask, request
from vettly import verify_webhook_signature, construct_webhook_event

app = Flask(__name__)
WEBHOOK_SECRET = "whsec_..."

@app.route("/webhooks/vettly", methods=["POST"])
def handle_webhook():
    payload = request.get_data(as_text=True)
    signature = request.headers.get("X-Vettly-Signature")

    if not verify_webhook_signature(payload, signature, WEBHOOK_SECRET):
        return "Invalid signature", 401

    event = construct_webhook_event(payload)

    if event["type"] == "decision.blocked":
        # Handle blocked content
        decision_id = event["data"]["decision_id"]
        notify_moderator(decision_id)
    elif event["type"] == "decision.flagged":
        # Queue for human review
        add_to_review_queue(event["data"])

    return "OK", 200
```

### Manage Webhooks

```python
# List all webhooks
webhooks = client.list_webhooks()

# Update a webhook
client.update_webhook(
    webhook_id="webhook-id",
    events=["decision.blocked"],
    enabled=True
)

# Test a webhook
test_result = client.test_webhook("webhook-id", "decision.blocked")
print(f"Test {'passed' if test_result['success'] else 'failed'}")

# View delivery logs
deliveries = client.get_webhook_deliveries("webhook-id", limit=10)

# Delete a webhook
client.delete_webhook("webhook-id")
```

---

## Idempotency

Prevent duplicate processing with request IDs:

```python
result = client.check(
    content="Hello",
    policy_id="default",
    request_id="unique-request-id-123"
)

# Same request_id returns cached result
duplicate = client.check(
    content="Hello",
    policy_id="default",
    request_id="unique-request-id-123"
)

assert result.id == duplicate.id  # Same decision
```

---

## Error Handling

The SDK provides typed exceptions for precise handling:

```python
from vettly import (
    VettlyAPIError,
    VettlyAuthError,
    VettlyRateLimitError,
    VettlyQuotaError,
    VettlyValidationError,
    VettlyServerError,
)

try:
    result = client.check(content="test", policy_id="default")
except VettlyAuthError:
    # Invalid or expired API key
    print("Authentication failed - check your API key")
except VettlyRateLimitError as e:
    # Rate limited - SDK retries automatically, this means retries exhausted
    print(f"Rate limited. Retry after {e.retry_after}s")
except VettlyQuotaError as e:
    # Monthly quota exceeded
    print(f"Quota exceeded: {e.quota}")
except VettlyValidationError as e:
    # Invalid request parameters
    print(f"Validation error: {e.errors}")
except VettlyServerError as e:
    # Server error (5xx)
    print(f"Server error: {e.status_code}")
except VettlyAPIError as e:
    # Other API errors
    print(f"{e.code}: {e.message}")
```

---

## Configuration

```python
from vettly import ModerationClient

client = ModerationClient(
    api_key="vettly_live_...",
    api_url="https://api.vettly.dev",  # Optional: custom API URL
    timeout=30.0,                       # Request timeout in seconds
    max_retries=3,                      # Max retries for failures
    retry_delay=1.0,                    # Base delay for exponential backoff
)
```

---

## Context Manager

Use as a context manager for automatic cleanup:

```python
# Sync client
with ModerationClient("vettly_live_...") as client:
    result = client.check(content="Hello", policy_id="default")

# Async client
async with AsyncModerationClient("vettly_live_...") as client:
    result = await client.check(content="Hello", policy_id="default")
```

---

## Response Models

### CheckResponse

```python
from vettly import Action, CheckResponse, CategoryResult

result: CheckResponse = client.check(...)

result.id           # str - Decision ID (UUID)
result.safe         # bool
result.flagged      # bool
result.action       # Action (ALLOW, WARN, FLAG, BLOCK)
result.categories   # List[CategoryResult]
result.latency_ms   # int
result.policy_id    # str
```

### CategoryResult

```python
for category in result.categories:
    category.category   # str - Category name
    category.score      # float - 0.0 to 1.0
    category.threshold  # float - Configured threshold
    category.triggered  # bool - Whether threshold was exceeded
```

### Action Enum

```python
from vettly import Action

Action.ALLOW   # Content passes all checks
Action.WARN    # Minor concern, user should be notified
Action.FLAG    # Needs human review
Action.BLOCK   # Violates policy
```

---

## Who Uses Vettly

### Trust & Safety Teams
- Audit trail for every decision
- Policy version history
- Evidence preservation for appeals

### Legal & Compliance
- DSA Article 17 compliance
- Decision records for legal discovery
- Policy approval workflows

### Engineering Teams
- Type-safe Pydantic models
- Automatic retries with exponential backoff
- Full async support

---

## Links

- [vettly.dev](https://vettly.dev) - Sign up
- [docs.vettly.dev](https://docs.vettly.dev) - Documentation
- [PyPI](https://pypi.org/project/vettly/) - Package