DEV Community

Rish
Rish

Posted on

Building Production-Ready SaaS Data Connectors & AI Safety Hooks in Python (With Checkpoint Pagination)

#ai #python #tutorial #webdev

Building reliable SaaS data connectors and secure AI agent tooling requires two critical components: robust checkpoint-based pagination for incremental syncs and strict pre-tool execution safety hooks. In this post, we explore how to implement both in Python.

1. SaaS REST Connector with Checkpoint Pagination & Exponential Backoff

When syncing data from APIs (like Jira, GitHub, or Salesforce), handling pagination and transient rate limits gracefully is essential. Here is a production-grade Python pattern using Pydantic v2 for state serialization:

import time
from typing import Dict, Any, Generator, Optional
from pydantic import BaseModel, Field

class ConnectorState(BaseModel):
    cursor: Optional[str] = None
    offset: int = 0
    last_sync_timestamp: float = 0.0

class SaaSRESTConnector:
    def __init__(self, api_key: str, base_url: str):
        self.api_key = api_key
        self.base_url = base_url
        self.state = ConnectorState()

    def fetch_page_with_backoff(self, endpoint: str, params: Dict[str, Any], max_retries: int = 5) -> Dict[str, Any]:
        """Fetch API pages with exponential backoff on failure."""
        for attempt in range(max_retries):
            try:
                # Simulated REST request
                return {"items": [{"id": 1}], "next_cursor": None}
            except Exception as e:
                if attempt == max_retries - 1:
                    raise e
                time.sleep(2 ** attempt)

    def load_from_state(self, state_dict: Dict[str, Any]):
        """Resume connector from saved checkpoint state."""
        self.state = ConnectorState(**state_dict)

    def stream_documents(self) -> Generator[Dict[str, Any], None, None]:
        """Stream documents iteratively while updating checkpoint state."""
        while True:
            params = {"cursor": self.state.cursor, "offset": self.state.offset}
            data = self.fetch_page_with_backoff("/documents", params)
            for item in data.get("items", []):
                yield item
            self.state.cursor = data.get("next_cursor")
            if not self.state.cursor:
                break
Enter fullscreen mode Exit fullscreen mode

2. AI Agent Safety Hook (PreToolUse Destructive Bash Blocker)

When deploying AI coding agents (like Claude Code), blocking dangerous bash commands before execution prevents accidental system or data loss:

import sys
import json
import logging

DESTRUCTIVE_PATTERNS = [
    "rm -rf", "drop database", "git reset --hard", 
    "mkfs", "dd if=", "chmod -r 777"
]

def audit_and_block(command: str) -> bool:
    for pattern in DESTRUCTIVE_PATTERNS:
        if pattern in command.lower():
            logging.error(f"BLOCKED DESTRUCTIVE BASH COMMAND: {command}")
            return True
    return False

if __name__ == "__main__":
    tool_input = json.loads(sys.stdin.read()) if not sys.stdin.isatty() else {}
    cmd = tool_input.get("command", "")
    if audit_and_block(cmd):
        print("Error: Destructive command blocked by safety hook.", file=sys.stderr)
        sys.exit(1)
    sys.exit(0)
Enter fullscreen mode Exit fullscreen mode

Production Developer Kit & Resources

Get the full boilerplate, hooks, test suites, and template documentation:

Top comments (0)