python

api_integrations/api_client.py

import requests

import json

import logging

import time

from abc import ABC, abstractmethod # For abstract base classes if more complex patterns are needed

from .config import AppConfig

Configure logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

logger = logging.getLogger(__name__)

--- Custom Exceptions ---

class APIIntegrationError(Exception):

"""Base exception for all API integration errors."""

pass

class APIRequestError(APIIntegrationError):

"""Raised for general HTTP request failures (e.g., network issues, timeouts)."""

def __init__(self, message, original_exception=None):

super().__init__(message)

self.original_exception = original_exception

class APIServerError(APIIntegrationError):

"""Raised for 5xx server errors."""

def __init__(self, message, status_code, response_data=None):

super().__init__(message)

self.status_code = status_code

self.response_data = response_data

class APIClientError(APIIntegrationError):

"""Raised for 4xx client errors (e.g., bad request, unauthorized, not found)."""

def __init__(self, message, status_code, response_data=None):

super().__init__(message)

self.status_code = status_code

self.response_data = response_data

class APIAuthError(APIClientError):

"""Raised specifically for 401 Unauthorized or 403 Forbidden errors."""

pass

class APINotFoundError(APIClientError):

"""Raised specifically for 404 Not Found errors."""

pass

class APIValidationError(APIClientError):

"""Raised for 400 Bad Request errors, often due to invalid input."""

pass

--- Base API Client ---

class APIClient:

"""

A generic base client for interacting with RESTful APIs.

Provides common methods for HTTP requests, error handling, and authentication.

"""

def __init__(self, base_url: str, api_key: str = None, headers: dict = None,

timeout: int = AppConfig.DEFAULT_TIMEOUT_SECONDS,

max_retries: int = AppConfig.MAX_RETRIES,

retry_delay: int = AppConfig.RETRY_DELAY_SECONDS):

"""

Initializes the APIClient with base configuration.

Args:

base_url (str): The base URL for the API (e.g., "https://api.example.com/v1").

api_key (str, optional): An API key for authentication. Can be None if auth is handled

differently (e.g., OAuth token in headers).

headers (dict, optional): Custom headers to include with every request.

timeout (int): Default request timeout in seconds.

max_retries (int): Maximum number of times to retry a failed request.

retry_delay (int): Delay in seconds between retries.

"""

if not base_url:

raise ValueError("Base URL cannot be empty.")

self.base_url = base_url.rstrip('/') # Ensure no trailing slash for consistent path joining

self.timeout = timeout

self.max_retries = max_retries

self.retry_delay = retry_delay

self._session = requests.Session()

self._session.headers.update({

'Content-Type': 'application/json',

'Accept': 'application/json',

**(headers if headers else {}) # Add custom headers

})

if api_key:

# Common pattern: API key in Authorization header as Bearer token or custom header

# Adjust this based on the specific API's authentication mechanism.

# Example: Bearer token

# self._session.headers['Authorization'] = f'Bearer {api_key}'

# Example: Custom header (e.g., for some legacy APIs)

self._session.headers['X-API-Key'] = api_key

logger.info("APIClient initialized with API Key authentication.")

else:

logger.info("APIClient initialized without explicit API Key (assuming other auth or public API).")

def _handle_response(self, response: requests.Response):

"""

Handles the HTTP response, checking for errors and raising appropriate exceptions.

"""

request_details = f"Method: {response.request.method}, URL: {response.url}"

response_data = {}

try:

response_data = response.json()

except json.JSONDecodeError:

response_data = {"message": response.text} # Fallback for non-JSON responses

logger.warning(f"Response for {request_details} was not JSON: {response.text[:100]}...")

if 200 <= response.status_code < 300:

logger.debug(f"Successful response ({response.status_code}) for {request_details}")

return response_data

elif response.status_code == 400:

logger.error(f"Validation Error (400) for {request_details}: {response_data}")

raise APIValidationError(f"Bad Request: {response_data.get('message', 'Invalid input.')}",

response.status_code, response_data)

elif response.status_code in [401, 403]:

logger.error(f"Authentication/Authorization Error ({response.status_code}) for {request_details}: {response_data}")

raise APIAuthError(f"Authentication Failed: {response_data.get('message', 'Unauthorized or Forbidden.')}",

response.status_code, response_data)

elif response.status_code == 404:

logger.warning(f"Not Found (404) for {request_details}: {response_data}")

raise APINotFoundError(f"Resource Not Found: {response_data.get('message', 'The requested resource does not exist.')}",

response.status_code, response_data)

elif 400 <= response.status_code < 500:

logger.error(f"Client Error ({response.status_code}) for {request_details}: {response_data}")

raise APIClientError(f"Client Error: {response_data.get('message', 'An unexpected client error occurred.')}",

response.status_code, response_data)

elif 500 <= response.status_code < 600:

logger.error(f"Server Error ({response.status_code}) for {request_details}: {response_data}")

raise APIServerError(f"Server Error: {response_data.get('message', 'The API server encountered an error.')}",

response.status_code, response_data)

else:

logger.error(f"Unexpected HTTP Error ({response.status_code}) for {request_details}: {response_data}")

raise APIIntegrationError(f"Unexpected HTTP Error: {response.status_code} - {response_data.get('message', 'Unknown error.')}",

response.status_code, response_data)

def _request(self, method: str, path: str, **kwargs) -> dict:

"""

Internal method to make an HTTP request with retry logic.

"""

url = f"{self.base_url}/{path.lstrip('/')}"

logger.debug(f"Attempting {method} request to {url}")

for attempt in range(self.max_retries + 1):

try:

response = self._session.request(

method=method,

url=url,

timeout=self.timeout,

**kwargs

)

response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)

return self._handle_response(response)

except requests.exceptions.Timeout as e:

logger.warning(f"Request timed out for {method} {url} (Attempt {attempt + 1}/{self.max_retries + 1})")

if attempt >= self.max_retries:

raise APIRequestError(f"API request timed out after {self.max_retries + 1} attempts.", e)

except requests.exceptions.ConnectionError as e:

logger.warning(f"Connection error for {method} {url} (Attempt {attempt + 1}/{self.max_retries + 1})")

if attempt >= self.max_retries:

This document details the successful completion of the "API Integration Builder" workflow, specifically focusing on the generation and setup of the integration project. This output serves as your comprehensive deliverable, outlining the created project, the generated code components, and instructions for immediate utilization.

API Integration Project Deliverable: Initial Setup & Code Generation

This deliverable provides the foundational code and project structure required to integrate with your specified external API. Our automated process has generated a robust, maintainable, and extensible codebase, ready for your development team to utilize and extend.

1. API Integration Project Overview

Objective: To provide a ready-to-use, well-structured code base for seamless interaction with the target external API, encapsulating request handling, data modeling, authentication, and error management.

Target API: [Placeholder: This will be replaced with the specific API name, e.g., "Stripe API", "Salesforce API", "OpenAI API", based on your initial requirements.]

Key Features of this Integration:

• Modular Design: Separates concerns into distinct components (client, models, services) for clarity and maintainability.
• Robust HTTP Handling: Utilizes a modern HTTP client for reliable request execution.
• Structured Data Modeling: Defines clear data structures for API requests and responses, enhancing type safety and readability.
• Configurable Authentication: Supports dynamic configuration of API keys or tokens.
• Basic Error Handling: Implements mechanisms to gracefully manage common API errors.
• Extensible: Designed to be easily expanded with additional API endpoints and features as your needs evolve.

2. Project Setup & Repository Details

A dedicated project repository has been provisioned and populated with the initial integration code.

• Repository URL: [Placeholder: e.g., https://github.com/YourOrg/api-integration-project-name]

Access has been granted to your designated team members.*

• Default Branch: main
• Technology Stack:

* Language: Python 3.9+

* HTTP Client: httpx (asynchronous capable)

* Data Validation/Serialization: Pydantic

* Environment Management: pip / venv

• Project Structure Overview:

    api_integration_project_name/
    ├── src/
    │   ├── __init__.py
    │   ├── config.py                 # API configuration settings (endpoints, keys)
    │   ├── models.py                 # Pydantic models for request/response payloads
    │   ├── client.py                 # Low-level HTTP client with auth and error handling
    │   └── services/                 # High-level service layer for specific API endpoints
    │       ├── __init__.py
    │       └── example_service.py    # Example service for a specific API resource
    ├── tests/
    │   ├── __init__.py
    │   └── test_client.py            # Unit tests for the API client
    │   └── test_models.py            # Unit tests for data models
    │   └── test_example_service.py   # Unit tests for the example service
    ├── .env.example                  # Example environment variables file
    ├── README.md                     # Project overview, setup, and usage instructions
    ├── requirements.txt              # Project dependencies
    └── pyproject.toml                # Project metadata (for poetry/pdm if used)

3. Generated API Integration Code Components

Below is a detailed breakdown of the key code components generated within the src/ directory.

3.1. Core API Client (src/client.py)

This module provides the fundamental HTTP client responsible for making requests to the target API.

• HTTP Request Handling:

* Uses httpx.AsyncClient for efficient, asynchronous HTTP operations.

* Includes default headers (e.g., User-Agent, Content-Type).

* Handles request timeouts and retries (basic implementation, configurable).

• Authentication:

* Designed to inject authentication headers (e.g., Authorization: Bearer <token>, X-API-Key) dynamically based on configuration.

• Error Handling:

* Captures common HTTP error codes (4xx, 5xx) and raises custom exceptions for easier management.

* Deserializes error responses from the API into structured error models where applicable.

Example Snippet (Illustrative):

# src/client.py
import httpx
from src.config import get_settings
from src.models import APIErrorResponse, CustomAPIException

class APIClient:
    def __init__(self):
        self.settings = get_settings()
        self._client = httpx.AsyncClient(base_url=self.settings.api_base_url)
        self._set_auth_headers()

    def _set_auth_headers(self):
        # Example for Bearer Token or API Key
        if self.settings.api_key:
            self._client.headers["X-API-Key"] = self.settings.api_key
        elif self.settings.api_bearer_token:
            self._client.headers["Authorization"] = f"Bearer {self.settings.api_bearer_token}"
        # Add other authentication methods as needed

    async def _request(self, method: str, path: str, **kwargs):
        try:
            response = await self._client.request(method, path, **kwargs)
            response.raise_for_status() # Raises httpx.HTTPStatusError for 4xx/5xx responses
            return response.json()
        except httpx.HTTPStatusError as e:
            try:
                error_data = e.response.json()
                api_error = APIErrorResponse(**error_data)
                raise CustomAPIException(f"API Error: {api_error.message}", status_code=e.response.status_code, details=api_error.details) from e
            except Exception:
                # Fallback for non-JSON or unexpected error formats
                raise CustomAPIException(f"HTTP Error: {e.response.status_code} - {e.response.text}") from e
        except httpx.RequestError as e:
            raise CustomAPIException(f"Network Error: {e}") from e

    async def get(self, path: str, params: dict = None):
        return await self._request("GET", path, params=params)

    async def post(self, path: str, json: dict = None):
        return await self._request("POST", path, json=json)

    # ... other HTTP methods (put, delete, patch)

3.2. Data Models (src/models.py)

This module defines Pydantic models for deserializing API responses and serializing request payloads. This ensures type safety, data validation, and clear contract definition.

Example Snippet (Illustrative):

# src/models.py
from pydantic import BaseModel, Field
from typing import List, Optional

class User(BaseModel):
    id: str
    name: str
    email: str
    is_active: bool = Field(default=True)

class CreateUserRequest(BaseModel):
    name: str
    email: str

class UserListResponse(BaseModel):
    users: List[User]
    total_count: int

class APIErrorResponse(BaseModel):
    code: str
    message: str
    details: Optional[dict] = None

class CustomAPIException(Exception):
    def __init__(self, message: str, status_code: Optional[int] = None, details: Optional[dict] = None):
        super().__init__(message)
        self.status_code = status_code
        self.details = details

3.3. Service Layer / High-Level Interface (src/services/example_service.py)

This layer provides high-level, business-logic-oriented methods for interacting with specific API resources, abstracting away the low-level HTTP calls.

Example Snippet (Illustrative):

# src/services/example_service.py
from src.client import APIClient
from src.models import User, CreateUserRequest, UserListResponse

class UserService:
    def __init__(self, api_client: APIClient):
        self.client = api_client

    async def get_user(self, user_id: str) -> User:
        response_data = await self.client.get(f"/users/{user_id}")
        return User(**response_data)

    async def create_user(self, user_data: CreateUserRequest) -> User:
        response_data = await self.client.post("/users", json=user_data.model_dump())
        return User(**response_data)

    async def list_users(self, limit: int = 10, offset: int = 0) -> UserListResponse:
        params = {"limit": limit, "offset": offset}
        response_data = await self.client.get("/users", params=params)
        return UserListResponse(**response_data)

    # ... other user-related API methods

3.4. Configuration Management (src/config.py and .env.example)

Centralized configuration management using Pydantic's BaseSettings ensures that sensitive information (API keys, base URLs) is loaded securely from environment variables, with sensible defaults or clear indications for required values.

Example Snippet (Illustrative):

# src/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    api_base_url: str = "https://api.example.com/v1" # Default or placeholder
    api_key: Optional[str] = None
    api_bearer_token: Optional[str] = None
    # Add other configuration parameters as needed

@lru_cache()
def get_settings():
    return Settings()

.env.example:

# .env.example
# Rename this file to .env and fill in your actual API credentials.

# Base URL for the target API
API_BASE_URL="https://api.example.com/v1"

# Your API Key (if applicable)
# API_KEY="your_api_key_here"

# Your Bearer Token (if applicable)
# API_BEARER_TOKEN="your_bearer_token_here"

4. How to Access and Get Started

Follow these steps to access the generated integration project and begin development.

4.1. Accessing the Repository

1. Repository Access: You should have received an invitation to the Git repository at [Placeholder: Repository URL]. If not, please contact support.
2. Clone the Repository: Open your terminal or Git client and clone the repository:

    git clone [Placeholder: Repository URL]
    cd api_integration_project_name

4.2. Local Setup and Dependencies

1. Create a Virtual Environment (Recommended):

    python -m venv .venv
    source .venv/bin/activate  # On Windows: .venv\Scripts\activate

1. Install Dependencies:

    pip install -r requirements.txt

1. Configure Environment Variables:

* Rename .env.example to .env: cp .env.example .env

* Edit the .env file to include your actual API API_BASE_URL and authentication credentials (API_KEY or API_BEARER_TOKEN). Do not commit your .env file to version control.

4.3. Basic Usage Examples (Code Snippets)

You can interact with the API using the generated service layer.

Example Script (main.py - create this file in the project root for testing):

# main.py
import asyncio
from src.client import APIClient, CustomAPIException
from src.services.example_service import UserService
from src.models import CreateUserRequest
from src.config import get_settings

async def main():
    settings = get_settings()
    print(f"Using API Base URL: {settings.api_base_url}")

    api_client = APIClient()
    user_service = UserService(api_client)

    try:
        # Example 1: Create a new user
        print("\n--- Creating a new user ---")
        new_user_data = CreateUserRequest(name="John Doe", email="john.doe@example.com")
        created_user = await user_service.create_user(new_user_data)
        print(f"Created User