Integration Patterns

import requests

BASE_URL = "https://api.advancehq.com"
API_KEY = "your_api_key"
headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

# Step 1: Create the insured (customer)
insured_data = {
    "insured_name": "ABC Manufacturing Inc",
    "email_primary": "[email protected]",
    "phone_number": "555-111-2222",
    "address1": "789 Industrial Blvd",
    "city": "Detroit",
    "address_state": "MI",
    "zip_code": "48201"
}

response = requests.post(f"{BASE_URL}/v1/insured", headers=headers, json=insured_data)
insured = response.json()
insured_id = insured["insured_id"]

# Step 2: Create the plan (invoice)
plan_data = {
    "insured_id": insured_id,
    "payment_due": "2025-02-28T00:00:00Z",
    "status": "open",
    "reminder_days": [-7, -1, 3],
    "policies": [
        {
            "policy_number": "GL-2025-001",
            "effective_date": "2025-02-01",
            "expiration_date": "2026-02-01",
            "policy_items": [
                {"name": "General Liability Premium", "amount": "8500.00"},
                {"name": "State Taxes", "amount": "425.00"}
            ]
        }
    ]
}

response = requests.post(f"{BASE_URL}/v1/plans", headers=headers, json=plan_data)
plan = response.json()

print(f"Invoice created: {plan['plan_number']}")
print(f"Payment URL: {plan['payment_url']}")
print(f"Invoice PDF: {plan['invoice_pdf_url']}")

# Step 3: Monitor payment status
plan_id = plan["plan_id"]
response = requests.get(f"{BASE_URL}/v1/plans/{plan_id}", headers=headers)
plan_details = response.json()

print(f"Status: {plan_details['status']}")
print(f"Collected: ${plan_details['collected_amount']} of ${plan_details['total_invoiced_amount']}")

import requests
from datetime import datetime, timedelta

BASE_URL = "https://api.advancehq.com"
API_KEY = "your_api_key"
headers = {"Authorization": f"Bearer {API_KEY}"}

def sync_settled_payments():
    """Fetch all settled payments for reconciliation"""
    all_payments = []
    page = 1
    
    while True:
        response = requests.get(
            f"{BASE_URL}/v1/payments",
            headers=headers,
            params={
                "pagination": "true",
                "page": page,
                "page_size": 100,
                "status": "Settled"
            }
        )
        data = response.json()
        all_payments.extend(data["data"])
        
        if not data["pagination"]["has_next"]:
            break
        page += 1
    
    return all_payments

# Get all settled payments
payments = sync_settled_payments()

# Process for your accounting system
for payment in payments:
    print(f"Payment {payment['payment_number']}: ${payment['txn_amount']} - {payment['settled_at']}")

import requests

BASE_URL = "https://api.advancehq.com"
API_KEY = "your_api_key"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Get all producers
response = requests.get(f"{BASE_URL}/v1/producers", headers=headers)
producers = response.json()

# Get plans with commissions for each producer
for producer in producers:
    producer_id = producer["producer_id"]
    
    # Get all paid plans for this producer
    response = requests.get(
        f"{BASE_URL}/v1/plans",
        headers=headers,
        params={"status": "paid"}
    )
    plans = response.json()
    
    # Filter plans for this producer and calculate commissions
    producer_plans = [p for p in plans if p.get("producer_id") == producer_id]
    total_commission = sum(float(p.get("commission_amount", 0)) for p in producer_plans)
    
    print(f"{producer['producer_name']}: ${total_commission:.2f} in commissions")

import requests
import time
from typing import Optional, Dict, Any

class AdvanceAPIClient:
    def __init__(self, api_key: str, base_url: str = "https://api.advancehq.com"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def _request(
        self, 
        method: str, 
        endpoint: str, 
        data: Optional[Dict] = None,
        params: Optional[Dict] = None,
        max_retries: int = 3
    ) -> Dict[Any, Any]:
        """Make API request with retry logic and error handling"""
        url = f"{self.base_url}{endpoint}"
        
        for attempt in range(max_retries):
            try:
                response = self.session.request(
                    method=method,
                    url=url,
                    json=data,
                    params=params
                )
                
                # Handle rate limiting
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 60))
                    if attempt < max_retries - 1:
                        time.sleep(retry_after)
                        continue
                    raise Exception("Rate limit exceeded after retries")
                
                # Handle success
                if response.status_code in [200, 201]:
                    return response.json()
                
                # Handle client errors
                if 400 <= response.status_code < 500:
                    error_detail = response.json().get("detail", "Unknown error")
                    raise Exception(f"API Error {response.status_code}: {error_detail}")
                
                # Handle server errors with retry
                if response.status_code >= 500:
                    if attempt < max_retries - 1:
                        time.sleep(2 ** attempt)  # Exponential backoff
                        continue
                    raise Exception(f"Server error {response.status_code}")
                    
            except requests.RequestException as e:
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)
                    continue
                raise e
        
        raise Exception("Max retries exceeded")
    
    def create_plan(self, plan_data: Dict) -> Dict:
        return self._request("POST", "/v1/plans", data=plan_data)
    
    def get_plan(self, plan_id: str) -> Dict:
        return self._request("GET", f"/v1/plans/{plan_id}")
    
    def list_payments(self, **params) -> Dict:
        return self._request("GET", "/v1/payments", params=params)

# Usage
client = AdvanceAPIClient("your_api_key")

try:
    plan = client.create_plan({
        "recipient_name": "Test Customer",
        "recipient_email": "[email protected]",
        "status": "open"
    })
    print(f"Created plan: {plan['plan_id']}")
except Exception as e:
    print(f"Error: {e}")