Workflow: API Integration Builder
Step: collab → generate_code
This document provides a comprehensive, detailed, and professional output for generating production-ready code to integrate with external APIs. As step 1 of the "API Integration Builder" workflow, this output focuses on establishing a robust, flexible, and maintainable foundation for your API interactions.
You are receiving the initial code generation for your API integration project. This output delivers a professional, well-structured, and highly adaptable Python API client template. While the specific external API you intend to integrate with has not yet been specified, this foundational code is designed to be easily customized for virtually any RESTful API, incorporating best practices for reliability, security, and maintainability.
This deliverable includes:
Our goal is to provide you with a powerful starting point that minimizes development time and ensures a high-quality integration.
The "API Integration Builder" workflow is designed to streamline the process of connecting your systems with external services.
* Generic API Client: A flexible Python class designed to handle various HTTP methods (GET, POST, PUT, DELETE).
* Robust Error Handling: Mechanisms to gracefully manage API errors, network issues, and unexpected responses.
* Retry Logic with Exponential Backoff: Automatic retries for transient errors to improve reliability.
* Configurable Authentication: Support for API keys, with clear pathways for OAuth2, Bearer tokens, etc.
* Logging Integration: Detailed logging for monitoring and debugging.
* Environment-based Configuration: Separation of sensitive data and settings from the codebase.
* Clear Structure & Comments: Code that is easy to understand, modify, and extend.
Before diving into the code, understanding the underlying principles is crucial for building reliable integrations:
* Error Handling: Anticipate and handle various error types (network, API-specific, data parsing).
* Retries: Implement intelligent retry mechanisms for transient failures.
* Timeouts: Prevent indefinite waits for unresponsive APIs.
* Secure Authentication: Use appropriate and secure authentication methods (OAuth2, API Keys via environment variables).
* Data Protection: Encrypt sensitive data in transit (HTTPS) and at rest.
* Input Validation: Sanitize and validate all data sent to and received from the API.
* Modularity: Organize code into logical, reusable components (e.g., an ApiClient class).
* Configuration: Externalize settings (base URLs, API keys) from the code.
* Readability: Use clear variable names, comments, and consistent formatting.
* Rate Limiting: Respect API rate limits to avoid being blocked.
* Asynchronous Operations: Consider async clients for high-throughput scenarios (beyond this initial template, but good to keep in mind).
* Efficient Resource Usage: Manage connections and memory effectively.
The proposed architecture centers around a dedicated ApiClient class that encapsulates all logic for interacting with a specific external API. This class will manage HTTP requests, authentication, error handling, retries, and data parsing, providing a clean interface for your application logic.
+---------------------+ +---------------------+ +---------------------+
| Your Application | | API Configuration | | External API |
| Logic |----->| (Environment Vars, |----->| (e.g., RESTful API) |
| (e.g., `main.py`) | | .env file) | | |
+---------------------+ +---------------------+ +---------------------+
| ^
| |
v |
+---------------------+ |
| `ApiClient` Class | |
| (`api_client.py`) | |
| - Handles requests | |
| - Authentication |------------------+
| - Error handling |
| - Retries |
| - Logging |
+---------------------+
|
v
+---------------------+
| Logging System |
| (e.g., Console, |
| File, Cloudwatch) |
+---------------------+
python
import requests
import logging
import time
from typing import Dict, Any, Optional
from config import API_BASE_URL, API_KEY, RETRY_ATTEMPTS, RETRY_DELAY_SECONDS, TIMEOUT_SECONDS
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class ApiClientError(Exception):
"""Base exception for API Client errors."""
pass
class ApiRequestError(ApiClientError):
"""Raised for HTTP errors (4xx or 5xx responses)."""
def __init__(self, message: str, status_code: int, response_data: Optional[Dict] = None):
super().__init__(message)
self.status_code = status_code
self.response_data = response_data
class ApiConnectionError(ApiClientError):
"""Raised for network or connection-related errors."""
pass
class ApiTimeoutError(ApiClientError):
"""Raised when an API request times out."""
pass
class ApiRetryExhaustedError(ApiClientError):
"""Raised when all retry attempts are exhausted."""
pass
class ApiDeserializationError(ApiClientError):
"""Raised when response content cannot be deserialized."""
pass
class ApiClient:
"""
A generic API client for interacting with external RESTful services.
This client handles HTTP requests, authentication, error handling,
retries with exponential backoff, and logging.
"""
def __init__(self, base_url: str, api_key: str):
"""
Initializes the ApiClient.
Args:
base_url (str): The base URL for the API (e.g., "https://api.example.com/v1").
api_key (str): The API key for authentication.
"""
if not base_url:
raise ValueError("API base URL cannot be empty.")
if not api_key:
logger.warning("API key is empty. Ensure this is intentional for unauthenticated endpoints.")
self.base_url = base_url.rstrip('/') # Ensure no trailing slash for consistent path joining
self.api_key = api_key
self.session = requests.Session() # Use a session for connection pooling and cookie persistence
logger.info(f"ApiClient initialized for base URL: {self.base_url}")
def _get_headers(self, custom_headers: Optional[Dict] = None) -> Dict:
"""
Constructs the default request headers, including authentication.
Args:
custom_headers (Optional[Dict]): Additional headers to merge.
Returns:
Dict: A dictionary of headers.
"""
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"Authorization": f"Bearer {self.api_key}" # Example: Bearer token. Adapt for other auth types.
}
if custom_headers:
headers.update(custom_headers)
return headers
def _handle_response(self, response: requests.Response) -> Any:
"""
Handles the API response, checking for errors and parsing JSON.
Args:
response (requests.Response): The response object from the requests library.
Returns:
Any: The JSON-parsed response data.
Raises:
ApiRequestError: If the response indicates an HTTP error (4xx or 5xx).
ApiDeserializationError: If the response content cannot be parsed as JSON.
"""
try:
response.raise_for_status() # Raises HTTPError for 4XX/5XX responses
return response.json()
except requests.exceptions.HTTPError as e:
status_code = response.status_code
response_data = None
try:
response_data = response.json()
except requests.exceptions.JSONDecodeError:
response_data = {"error": response.text} # Fallback to raw text if not JSON
error_message = f"API request failed with status {status_code}: {response_data}"
logger.error(error_message)
raise ApiRequestError(error_message, status_code, response_data)
except requests.exceptions.JSONDecodeError as e:
error_message = f"Failed to deserialize JSON response from {response.url}: {e}"
logger.error(error_message)
raise ApiDeserializationError(error_message)
except Exception as e:
error_message = f"An unexpected error occurred while handling response: {e}"
logger.error(error_message)
raise ApiClientError(error_message)
def _make_request(
self,
method: str,
path: str,
params: Optional[Dict] = None,
json_data: Optional[Dict] = None,
data: Optional[Any] = None,
custom_headers: Optional[Dict] = None,
stream: bool = False
) -> Any:
"""
Internal method to make an HTTP request with retry logic.
Args:
method (str): The HTTP method (e.g., 'GET', 'POST').
path (str): The API endpoint path (e.g., "/users").
params (Optional[Dict]): Dictionary of query parameters.
json_data (Optional[Dict]): Dictionary of JSON data for the request body.
data (Optional[Any]): Data for the request body (e.g., form data, raw text).
custom_headers (Optional[Dict]): Additional headers for this specific request.
stream (bool): Whether to stream the response content.
Returns:
Any: The JSON-parsed response data.
Raises:
ApiConnectionError: For network connectivity issues.
ApiTimeoutError: If the request times out.
ApiRetryExhaustedError: If all retry attempts fail.
ApiClientError: For other unexpected client-side errors.
"""
url = f"{self.base_url}/{path.lstrip('/')}"
headers = self._get_headers(custom_headers)
current_attempt = 0
while current_attempt < RETRY_ATTEMPTS:
current_attempt += 1
try:
logger.debug(f"Attempt {current_attempt}/{RETRY_ATTEMPTS}: {method} {url}")
response = self.session.request(
method=method,
url=url,
params=params,
json=json_data,
data=data,
headers=headers,
timeout=TIMEOUT_SECONDS,
stream=stream
)
# If a successful response (2xx) or a client error (4
This document outlines the successful execution of Step 2 of 2 in the "API Integration Builder" workflow, focusing on the crucial "Project Creation" phase. This step establishes the foundational structure and initial parameters for your API integration project, paving the way for efficient and tailored code generation.
The API Integration Builder workflow is designed to streamline and automate the generation of code necessary to integrate with external APIs. Its primary goal is to accelerate development cycles, reduce manual coding effort, and ensure best practices in API connectivity. This comprehensive process guides you from initial project definition through to actionable, deployable code.
The overarching objective of this workflow is to generate robust, maintainable, and efficient code that facilitates seamless interaction with your chosen external APIs. This includes handling various aspects such as authentication, request/response serialization, error handling, and data mapping, tailored to your specific requirements.
projectmanager → create_project)This step marks the official initiation of your API integration project within our system. It involves defining the core characteristics and requirements that will guide the subsequent stages of code generation and integration development.
The "create_project" phase is critical for:
Based on your input and our system's capabilities, the following critical details are established during this step:
CRM-Salesforce-Integration, PaymentGateway-Stripe-Webhook).* Base URL(s): The root endpoint(s) for the API(s).
* Primary API Documentation Link(s): References to official API documentation for detailed specifications.
Upon completion of the "Project Creation" step, the following outputs are generated and stored:
With the project successfully created, the workflow will now proceed to the detailed design and code generation phases:
This meticulous project creation process ensures:
To proceed with the next steps in the API Integration Builder workflow, we kindly request you to:
We are confident that this structured approach will lead to a highly efficient and successful API integration. Please do not hesitate to reach out if you have any questions or require further clarification.