API Integration

import os
import time
import logging
from typing import Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logger = logging.getLogger(__name__)

class StripeClient:
    """Production-ready Stripe API client with retries and error handling."""

    BASE_URL = "https://api.stripe.com/v1"

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.environ["STRIPE_SECRET_KEY"]
        self.session = self._build_session()

    def _build_session(self) -> requests.Session:
        session = requests.Session()
        session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/x-www-form-urlencoded",
            "Stripe-Version": "2024-06-20",
        })
        retry_strategy = Retry(
            total=4,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET", "POST", "DELETE"],
            respect_retry_after_header=True,
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("https://", adapter)
        return session

    def create_customer(self, email: str, name: str, metadata: Optional[dict] = None) -> dict:
        """Create a Stripe customer."""
        payload = {"email": email, "name": name}
        if metadata:
            for key, value in metadata.items():
                payload[f"metadata[{key}]"] = value
        response = self._request("POST", "/customers", data=payload)
        return response

    def create_payment_intent(self, amount_cents: int, currency: str = "usd",
                              customer_id: Optional[str] = None) -> dict:
        """Create a payment intent for a given amount."""
        payload = {"amount": amount_cents, "currency": currency}
        if customer_id:
            payload["customer"] = customer_id
        return self._request("POST", "/payment_intents", data=payload)

    def list_charges(self, customer_id: str, limit: int = 10) -> list:
        """List charges for a customer with automatic pagination."""
        charges = []
        params = {"customer": customer_id, "limit": limit}
        while True:
            data = self._request("GET", "/charges", params=params)
            charges.extend(data["data"])
            if not data["has_more"]:
                break
            params["starting_after"] = data["data"][-1]["id"]
        return charges

    def _request(self, method: str, path: str, **kwargs) -> dict:
        url = f"{self.BASE_URL}{path}"
        try:
            resp = self.session.request(method, url, **kwargs)
            resp.raise_for_status()
            return resp.json()
        except requests.exceptions.HTTPError as e:
            error_body = e.response.json().get("error", {})
            logger.error(
                "Stripe API error: type=%s code=%s message=%s",
                error_body.get("type"),
                error_body.get("code"),
                error_body.get("message"),
            )
            if e.response.status_code == 429:
                retry_after = int(e.response.headers.get("Retry-After", 5))
                logger.warning("Rate limited. Retry after %ds", retry_after)
            raise
        except requests.exceptions.ConnectionError:
            logger.error("Connection failed to Stripe API")
            raise

# Usage
client = StripeClient()
customer = client.create_customer("[email protected]", "Alice Smith")
payment = client.create_payment_intent(2500, customer_id=customer["id"])
print(f"Payment intent {payment['id']} for ${payment['amount']/100:.2f}")

import requests
from typing import Any, Optional

class GraphQLClient:
    """Lightweight GraphQL client with error handling and variable support."""

    def __init__(self, endpoint: str, headers: Optional[dict] = None):
        self.endpoint = endpoint
        self.session = requests.Session()
        if headers:
            self.session.headers.update(headers)

    def execute(self, query: str, variables: Optional[dict] = None) -> dict:
        """Execute a GraphQL query or mutation."""
        payload = {"query": query}
        if variables:
            payload["variables"] = variables

        response = self.session.post(self.endpoint, json=payload, timeout=30)
        response.raise_for_status()
        result = response.json()

        if "errors" in result:
            error_messages = [e["message"] for e in result["errors"]]
            raise GraphQLError(error_messages, result.get("data"))

        return result["data"]

class GraphQLError(Exception):
    def __init__(self, messages: list, partial_data: Any = None):
        self.messages = messages
        self.partial_data = partial_data
        super().__init__(f"GraphQL errors: {'; '.join(messages)}")

# Usage: Query a GitHub-style GraphQL API
client = GraphQLClient(
    endpoint="https://api.github.com/graphql",
    headers={"Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}"},
)

# Fetch repositories with pagination
query = """
query($owner: String!, $cursor: String) {
  user(login: $owner) {
    repositories(first: 10, after: $cursor, orderBy: {field: UPDATED_AT, direction: DESC}) {
      pageInfo {
        hasNextPage
        endCursor
      }
      nodes {
        name
        description
        stargazerCount
        primaryLanguage { name }
      }
    }
  }
}
"""

repos = []
cursor = None
while True:
    data = client.execute(query, {"owner": "octocat", "cursor": cursor})
    page = data["user"]["repositories"]
    repos.extend(page["nodes"])
    if not page["pageInfo"]["hasNextPage"]:
        break
    cursor = page["pageInfo"]["endCursor"]

for repo in repos:
    lang = repo["primaryLanguage"]["name"] if repo["primaryLanguage"] else "N/A"
    print(f"{repo['name']} ({lang}) - {repo['stargazerCount']} stars")