Python

Integrate GuardCrow into your Python applications using requests or httpx.

Basic Example

A simple example using the requests library:

analyze.pypython

import os
import requests

def analyze_content(content: str) -> dict:
    """Analyze content for toxicity and sentiment."""
    response = requests.post(
        "https://api.guardcrow.com/v1/analyze",
        headers={
            "X-API-Key": os.environ["GUARDCROW_API_KEY"],
            "Content-Type": "application/json",
        },
        json={"content": content},
    )
    response.raise_for_status()
    return response.json()


# Usage
result = analyze_content("Hello, this is a test message")
print(f"Passed: {result['passed']}, Score: {result['score']}")

Client Class with Retry Logic

A complete client implementation with error handling and automatic retries:

guardcrow.pypython

import os
import time
from typing import Optional
from dataclasses import dataclass
import requests
from requests.exceptions import RequestException


@dataclass
class AnalysisResult:
    passed: bool
    score: int
    sentiment: str
    description: str
    categories: list[str]
    processing_time_ms: int


class GuardCrowError(Exception):
    """Base exception for GuardCrow errors."""
    def __init__(self, code: str, message: str, status: int):
        self.code = code
        self.message = message
        self.status = status
        super().__init__(message)


class RateLimitError(GuardCrowError):
    """Raised when rate limit is exceeded."""
    def __init__(self, message: str, reset_time: Optional[int] = None):
        super().__init__("RATE_LIMIT_EXCEEDED", message, 429)
        self.reset_time = reset_time


class GuardCrowClient:
    """Client for the GuardCrow API."""

    def __init__(
        self,
        api_key: Optional[str] = None,
        base_url: str = "https://api.guardcrow.com/v1",
        max_retries: int = 3,
        timeout: int = 30,
    ):
        self.api_key = api_key or os.environ.get("GUARDCROW_API_KEY")
        if not self.api_key:
            raise ValueError("API key is required")

        self.base_url = base_url
        self.max_retries = max_retries
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            "X-API-Key": self.api_key,
            "Content-Type": "application/json",
        })

    def analyze(
        self,
        content: str,
        metadata: Optional[dict] = None,
    ) -> AnalysisResult:
        """
        Analyze content for toxicity and sentiment.

        Args:
            content: Text to analyze (max 10,000 characters)
            metadata: Optional metadata (source, user_id, etc.)

        Returns:
            AnalysisResult with moderation details

        Raises:
            GuardCrowError: On API errors
            RateLimitError: When rate limited
        """
        payload = {"content": content}
        if metadata:
            payload["metadata"] = metadata

        last_error = None

        for attempt in range(self.max_retries):
            try:
                response = self.session.post(
                    f"{self.base_url}/analyze",
                    json=payload,
                    timeout=self.timeout,
                )

                # Handle rate limiting
                if response.status_code == 429:
                    reset_time = response.headers.get("X-RateLimit-Reset")
                    wait_seconds = (
                        int(reset_time) - int(time.time())
                        if reset_time
                        else 2 ** attempt
                    )
                    if attempt < self.max_retries - 1:
                        time.sleep(max(wait_seconds, 1))
                        continue
                    raise RateLimitError(
                        "Rate limit exceeded",
                        int(reset_time) if reset_time else None,
                    )

                # Handle other errors
                if not response.ok:
                    error_data = response.json().get("error", {})
                    raise GuardCrowError(
                        code=error_data.get("code", "UNKNOWN"),
                        message=error_data.get("message", "Unknown error"),
                        status=response.status_code,
                    )

                data = response.json()
                return AnalysisResult(
                    passed=data["passed"],
                    score=data["score"],
                    sentiment=data["sentiment"],
                    description=data["description"],
                    categories=data["categories"],
                    processing_time_ms=data["processingTimeMs"],
                )

            except RequestException as e:
                last_error = e
                if attempt < self.max_retries - 1:
                    time.sleep(2 ** attempt)

        raise last_error or GuardCrowError(
            "NETWORK_ERROR", "Request failed", 0
        )

    def health_check(self) -> dict:
        """Check API health status."""
        response = self.session.get(
            f"{self.base_url}/health",
            timeout=self.timeout,
        )
        response.raise_for_status()
        return response.json()


# Usage
if __name__ == "__main__":
    client = GuardCrowClient()

    try:
        result = client.analyze(
            "This is a test message",
            metadata={"source": "user-comments"},
        )
        print(f"Passed: {result.passed}")
        print(f"Score: {result.score}")
        print(f"Sentiment: {result.sentiment}")
    except RateLimitError as e:
        print(f"Rate limited. Reset at: {e.reset_time}")
    except GuardCrowError as e:
        print(f"API error: {e.code} - {e.message}")

Async Version (httpx)

For async applications using httpx:

guardcrow_async.pypython

import os
import asyncio
from typing import Optional
import httpx


class AsyncGuardCrowClient:
    """Async client for the GuardCrow API."""

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.environ.get("GUARDCROW_API_KEY")
        self.base_url = "https://api.guardcrow.com/v1"

    async def analyze(self, content: str) -> dict:
        """Analyze content asynchronously."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/analyze",
                headers={
                    "X-API-Key": self.api_key,
                    "Content-Type": "application/json",
                },
                json={"content": content},
                timeout=30.0,
            )
            response.raise_for_status()
            return response.json()

    async def analyze_batch(self, contents: list[str]) -> list[dict]:
        """Analyze multiple contents concurrently."""
        tasks = [self.analyze(content) for content in contents]
        return await asyncio.gather(*tasks, return_exceptions=True)


# Usage
async def main():
    client = AsyncGuardCrowClient()

    # Single analysis
    result = await client.analyze("Test message")
    print(result)

    # Batch analysis
    contents = [
        "First message to check",
        "Second message to check",
        "Third message to check",
    ]
    results = await client.analyze_batch(contents)
    for content, result in zip(contents, results):
        if isinstance(result, Exception):
            print(f"Error analyzing: {content}")
        else:
            print(f"{content[:30]}... -> Score: {result['score']}")


asyncio.run(main())

FastAPI Integration

Create a moderation middleware for FastAPI:

main.pypython

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
import httpx
import os

app = FastAPI()

GUARDCROW_API_KEY = os.environ["GUARDCROW_API_KEY"]


class Comment(BaseModel):
    content: str
    user_id: str


class ModerationResult(BaseModel):
    passed: bool
    score: int
    sentiment: str


async def moderate_content(content: str) -> ModerationResult:
    """Call GuardCrow API to moderate content."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.guardcrow.com/v1/analyze",
            headers={
                "X-API-Key": GUARDCROW_API_KEY,
                "Content-Type": "application/json",
            },
            json={"content": content},
            timeout=30.0,
        )

        if response.status_code == 429:
            raise HTTPException(
                status_code=503,
                detail="Service temporarily unavailable",
            )

        response.raise_for_status()
        data = response.json()

        return ModerationResult(
            passed=data["passed"],
            score=data["score"],
            sentiment=data["sentiment"],
        )


@app.post("/comments")
async def create_comment(comment: Comment):
    # Moderate the content
    moderation = await moderate_content(comment.content)

    if not moderation.passed:
        raise HTTPException(
            status_code=400,
            detail=f"Content blocked: {moderation.sentiment}",
        )

    # Save the comment (your logic here)
    return {
        "status": "created",
        "moderation_score": moderation.score,
    }


@app.post("/moderate")
async def moderate_endpoint(comment: Comment):
    """Public moderation endpoint."""
    result = await moderate_content(comment.content)
    return result

Django Integration

Create a reusable validator for Django models:

validators.pypython

import os
import requests
from django.core.exceptions import ValidationError


def validate_content_safety(content: str, max_score: int = 6) -> None:
    """
    Django validator for content moderation.

    Usage in a model:
        content = models.TextField(validators=[validate_content_safety])
    """
    try:
        response = requests.post(
            "https://api.guardcrow.com/v1/analyze",
            headers={
                "X-API-Key": os.environ["GUARDCROW_API_KEY"],
                "Content-Type": "application/json",
            },
            json={"content": content},
            timeout=30,
        )
        response.raise_for_status()
        result = response.json()

        if result["score"] > max_score:
            raise ValidationError(
                f"Content flagged: {result['description']}",
                code="content_unsafe",
                params={"score": result["score"]},
            )
    except requests.RequestException:
        # Don't block on API errors
        pass


# Usage in models.py
from django.db import models
from .validators import validate_content_safety


class Comment(models.Model):
    content = models.TextField(
        validators=[validate_content_safety]
    )
    user = models.ForeignKey("User", on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)

Installation & Setup

Install the required dependencies:

Terminalbash

# Using pip
pip install requests

# Or for async support
pip install httpx

Set your API key as an environment variable:

.envbash

GUARDCROW_API_KEY=sk_live_your_api_key_here