Workflow: API Integration Builder
Step: collab → generate_code
Description: Generate code to integrate with external APIs
This deliverable provides comprehensive, production-ready Python code for integrating with a generic RESTful API. Given the general nature of the request ("API Integration Builder"), this output focuses on establishing a robust, flexible, and secure foundation for API communication. It includes best practices for authentication, error handling, configuration management, and clear code structure, designed to be easily adaptable to various external APIs.
The generated code serves as a detailed template, demonstrating how to interact with common API patterns (e.g., GET, POST, PUT, DELETE requests) and incorporates two prevalent authentication methods: API Key and OAuth 2.0 Client Credentials.
config.py file, promoting security and ease of deployment.Before running the generated code, ensure you have the following installed:
requests library: This library is used for making HTTP requests.You can install it using pip:
**Actionable Steps for Configuration:**
1. **Create `config.py`**: Create a file named `config.py` and paste the content above into it.
2. **Update `BASE_URL`**: Replace `"https://api.example.com/v1"` with the actual base URL of the API you are integrating.
3. **Choose Authentication Method**:
* **API Key**: Uncomment and populate `API_KEY` and `API_KEY_HEADER_NAME` with your actual API key and the correct header name (e.g., `Authorization`, `X-API-Key`).
* **OAuth 2.0 Client Credentials**: Uncomment and populate `OAUTH_TOKEN_URL`, `OAUTH_CLIENT_ID`, `OAUTH_CLIENT_SECRET`, and `OAUTH_SCOPES`.
* **Important**: For production environments, **always use environment variables** (`os.getenv`) to store sensitive credentials instead of hardcoding them directly in `config.py`. The provided template already uses `os.getenv` as a best practice.
4. **Review Other Settings**: Adjust `REQUEST_TIMEOUT`, `RETRY_ATTEMPTS`, `RETRY_DELAY_SECONDS`, and `LOG_LEVEL` as needed.
---
### 5. Code Structure Explanation
The generated code consists of two main parts:
1. **`api_client.py`**: This file contains the core `APIClient` class, which handles all communication with the external API. It manages authentication, request construction, error handling, and response parsing.
2. **`example_usage.py`**: This file demonstrates how to instantiate and use the `APIClient` to perform common operations like listing, creating, updating, and deleting resources.
---
### 6. Generated Code
#### 6.1. `api_client.py`
This is the main API client class. It's designed to be robust and adaptable.
python
import requests
import time
import json
import logging
from datetime import datetime, timedelta
from urllib.parse import urljoin
from config import APIConfig
logging.basicConfig(level=APIConfig.LOG_LEVEL,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class APIError(Exception):
"""Custom exception for API errors."""
def __init__(self, message, status_code=None, details=None):
super().__init__(message)
self.status_code = status_code
self.details = details
class APIClient:
"""
A robust client for interacting with a RESTful API.
Supports API Key and OAuth 2.0 Client Credentials authentication.
"""
def __init__(self):
self.base_url = APIConfig.BASE_URL
self.session = requests.Session()
self.timeout = APIConfig.REQUEST_TIMEOUT
self.retry_attempts = APIConfig.RETRY_ATTEMPTS
self.retry_delay = APIConfig.RETRY_DELAY_SECONDS
# OAuth 2.0 specific attributes
self._access_token = None
self._token_expires_at = None
self._oauth_token_url = getattr(APIConfig, 'OAUTH_TOKEN_URL', None)
self._oauth_client_id = getattr(APIConfig, 'OAUTH_CLIENT_ID', None)
self._oauth_client_secret = getattr(APIConfig, 'OAUTH_CLIENT_SECRET', None)
self._oauth_scopes = getattr(APIConfig, 'OAUTH_SCOPES', None)
# API Key specific attributes
self._api_key = getattr(APIConfig, 'API_KEY', None)
self._api_key_header_name = getattr(APIConfig, 'API_KEY_HEADER_NAME', 'X-API-Key')
self._configure_authentication()
def _configure_authentication(self):
"""Determines and configures the authentication method."""
if self._api_key:
logger.info("Using API Key authentication.")
# API Key is typically added to headers on each request
self._auth_method = "api_key"
elif self._oauth_token_url and self._oauth_client_id and self._oauth_client_secret:
logger.info("Using OAuth 2.0 Client Credentials authentication.")
self._auth_method = "oauth2"
self._get_oauth_token() # Get initial token
else:
logger.warning("No valid authentication method configured. Proceeding without authentication.")
self._auth_method = "none"
def _get_oauth_token(self):
"""
Requests a new OAuth 2.0 access token using client credentials flow.
"""
if not self._oauth_token_url or not self._oauth_client_id or not self._oauth_client_secret:
raise APIError("OAuth 2.0 credentials or token URL are not configured.")
logger.info("Attempting to obtain new OAuth 2.0 access token...")
try:
response = requests.post(
self._oauth_token_url,
data={
"grant_type": "client_credentials",
"client_id": self._oauth_client_id,
"client_secret": self._oauth_client_secret,
"scope": self._oauth_scopes
},
timeout=self.timeout
)
response.raise_for_status() # Raise an exception for HTTP errors (4xx or 5xx)
token_data = response.json()
self._access_token = token_data.get("access_token")
expires_in = token_data.get("expires_in", 3600) # Default to 1 hour if not provided
self._token_expires_at = datetime.now() + timedelta(seconds=expires_in - 60) # Refresh 60s early
if not self._access_token:
raise APIError("OAuth token response did not contain 'access_token'.",
status_code=response.status_code, details=token_data)
logger.info("Successfully obtained OAuth 2.0 access token.")
except requests.exceptions.RequestException as e:
logger.error(f"Failed to obtain OAuth 2.0 token: {e}")
raise APIError(f"Failed to obtain OAuth 2.0 token: {e}", details=str(e))
except json.JSONDecodeError as e:
logger.error(f"Failed to decode OAuth 2.0 token response: {e}")
raise APIError(f"Failed to decode OAuth 2.0 token response: {e}", details=str(e))
def _prepare_headers(self, custom_headers=None):
"""
Prepares request headers, including authentication.
"""
headers = {
"Content-Type": "application/json",
"Accept": "application/json"
}
if self._auth_method == "api_key":
if self._api_key:
headers[self._api_key_header_name] = self._api_key
else:
logger.warning("API Key is configured but not set.")
elif self._auth_method == "oauth2":
if not self._access_token or (self._token_expires_at and datetime.now() >= self._token_expires_at):
logger.info("OAuth token expired or not present. Refreshing token.")
self._get_oauth_token()
if self._access_token:
headers["Authorization"] = f"Bearer {self._access_token}"
else:
logger.warning("OAuth 2.0 is configured but no access token is available.")
if custom_headers:
headers.update(custom_headers)
return headers
def _request(self, method, path, params=None, data=None, json_data=None, headers=None):
"""
Internal method to make an HTTP request with retry logic.
"""
url = urljoin(self.base_url, path)
request_headers = self._prepare_headers(headers)
for attempt in range(self.retry_attempts):
try:
logger.debug(f"Attempt {attempt + 1}/{self.retry_attempts}: {method} {url}")
response = self.session.request(
method,
url,
params=params,
data=data,
json=json_data,
headers=request_headers,
timeout=self.timeout
)
response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
return response
except requests.exceptions.HTTPError as e:
logger.error(f"HTTP Error for {method} {url}: {e.response.status_code} - {e.response.text}")
if 400 <= e.response.status_code < 500 and e.response.status_code != 429: # Client error (not rate limit)
raise APIError(f"API client error: {e}", status_code=e.response.status_code, details=e.response.text)
if attempt == self.retry_attempts - 1: # Last attempt, re-raise
raise APIError(f"API server error after {self.retry_attempts} attempts: {e}",
status_code=e.response.status_code, details=e.response.text)
except requests.exceptions.Timeout as e:
logger.warning(f"Request timed out for {method} {url}: {e}")
if attempt == self.retry_attempts - 1:
raise APIError(f"Request timed out after {self.retry_attempts} attempts: {e}", details=
Workflow Step: projectmanager → create_project
Description: Generate comprehensive guidance and initial frameworks for setting up a new project focused on external API integration. This deliverable outlines the critical planning, architectural, and foundational code considerations to ensure a robust and scalable integration.
Welcome to the "API Integration Builder" workflow. This step, "projectmanager → create_project," is designed to provide you with a detailed and professional framework for initiating your API integration project. Successful API integration begins with meticulous planning and a well-structured project foundation. This output will guide you through the essential considerations, architectural decisions, and initial code structuring required to connect your systems with external services efficiently and securely.
Our goal is to empower you to create a project that is not only functional but also maintainable, scalable, and resilient to potential issues.
This section provides a structured guide for setting up your API integration project. Each point represents a critical consideration or task.
Before writing any code, clearly articulate what you aim to achieve with this integration.
* What data needs to be exchanged?
* What operations (read, write, update, delete) are required?
* What is the expected volume and frequency of API calls?
* What are the performance requirements (e.g., latency, throughput)?
Thoroughly research and understand the external API you intend to integrate with.
* Endpoints: Identify all relevant endpoints and their purpose.
* Authentication: Understand the required authentication method (API keys, OAuth2, JWT, basic auth).
* Request/Response Formats: Note data structures (JSON, XML), required headers, and expected response codes.
* Rate Limits & Quotas: Be aware of any restrictions on the number of requests per unit of time to avoid service interruptions.
* Error Codes: Understand the API's error handling and common error codes.
* Terms of Service: Review the API provider's terms to ensure compliance and understand usage restrictions.
Sketch out a high-level design of how your system will interact with the external API.
* Direct Integration: Your application calls the API directly.
* Middleware/Proxy: Use an intermediate service (e.g., API Gateway, integration platform) to handle requests, transformations, and security.
* Event-Driven: Use webhooks or message queues for asynchronous communication.
Prepare your development and deployment environments.
* Install necessary IDEs, programming language runtimes, and package managers.
* Set up a dedicated repository for the integration project using Version Control (e.g., Git).
* Configure .gitignore to exclude sensitive files and temporary artifacts.
* Never hardcode credentials. Use environment variables, a dedicated secrets management service (e.g., AWS Secrets Manager, HashiCorp Vault), or a .env file for local development.
* Ensure different credentials are used for development, staging, and production environments.
Implement secure methods for authenticating with the external API.
Define how data will be translated between your system and the external API.
Design for robust error handling to prevent integration failures from impacting your application.
Ensure visibility into the integration's operation and performance.
* Log all API requests and responses (at appropriate levels, e.g., debug for full payloads, info for summaries).
* Log authentication attempts and failures.
* Record error details, timestamps, and relevant context (e.g., user ID, request ID).
* Set up dashboards to track key metrics (e.g., API call volume, success rates, average response times, error rates).
* Configure alerts for critical failures, high error rates, or performance degradation.
Prioritize security throughout the integration lifecycle.
Plan how the integration will be deployed and managed in production.
Based on the above project setup and planning, the "API Integration Builder" would generate foundational code. Below is a conceptual example of a basic API client structure, demonstrating the type of boilerplate code you would receive, typically in a language like Python or Node.js.
This framework provides a starting point, encapsulating common patterns for API interaction, authentication, and error handling.
Example: Python API Client Structure
# api_client.py
import os
import requests
import json
import time
from requests.exceptions import RequestException, HTTPError
class MyApiClient:
"""
A client for interacting with the External API.
"""
# --- Configuration ---
BASE_URL = os.getenv("EXTERNAL_API_BASE_URL", "https://api.example.com/v1")
API_KEY = os.getenv("EXTERNAL_API_KEY") # Securely loaded from environment variables
TIMEOUT_SECONDS = 30
MAX_RETRIES = 3
RETRY_DELAY_SECONDS = 1 # Initial delay, will use exponential backoff
def __init__(self):
if not self.API_KEY:
raise ValueError("API_KEY environment variable not set.")
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {self.API_KEY}", # Example: Bearer token auth
"Content-Type": "application/json",
"Accept": "application/json"
})
print(f"API Client initialized for {self.BASE_URL}")
def _make_request(self, method, endpoint, params=None, data=None, json_data=None, attempts=0):
"""
Internal method to handle API requests with retry logic and error handling.
"""
url = f"{self.BASE_URL}/{endpoint}"
try:
print(f"Making {method} request to: {url}")
response = self.session.request(
method,
url,
params=params,
data=data,
json=json_data,
timeout=self.TIMEOUT_SECONDS
)
response.raise_for_status() # Raises HTTPError for bad responses (4xx or 5xx)
return response.json()
except HTTPError as e:
status_code = e.response.status_code
print(f"HTTP Error {status_code} for {url}: {e.response.text}")
# Implement retry logic for specific transient errors (e.g., 429, 5xx)
if status_code in [429, 500, 502, 503, 504] and attempts < self.MAX_RETRIES:
wait_time = self.RETRY_DELAY_SECONDS * (2 ** attempts) # Exponential backoff
print(f"Retrying in {wait_time} seconds (attempt {attempts + 1}/{self.MAX_RETRIES})...")
time.sleep(wait_time)
return self._make_request(method, endpoint, params, data, json_data, attempts + 1)
# Re-raise non-retryable errors or after max retries
raise ValueError(f"API Error {status_code}: {e.response.text}") from e
except RequestException as e:
print(f"Network or connection error for {url}: {e}")
raise ConnectionError(f"Could not connect to API: {e}") from e
except json.JSONDecodeError:
print(f"Failed to decode JSON from response for {url}: {response.text}")
raise ValueError(f"Invalid JSON response: {response.text}")
except Exception as e:
print(f"An unexpected error occurred for {url}: {e}")
raise
def get_resource(self, resource_id, query_params=None):
"""
Example: Fetches a specific resource by ID.
"""
endpoint = f"resources/{resource_id}"
return self._make_request("GET", endpoint, params=query_params)
def create_resource(self, payload):
"""
Example: Creates a new resource.
"""
endpoint = "resources"
return self._make_request("POST", endpoint, json_data=payload)
def update_resource(self, resource_id, payload):
"""
Example: Updates an existing resource.
"""
endpoint = f"resources/{resource_id}"
return self._make_request("PUT", endpoint, json_data=payload)
# --- Usage Example (in a separate script or main function) ---
if __name__ == "__main__":
# Ensure environment variables are set for this example
# Example: export EXTERNAL_API_BASE_URL="https://jsonplaceholder.typicode.com"
# Example: export EXTERNAL_API_KEY="YOUR_MOCK_API_KEY" (for APIs requiring one)
try:
client = MyApiClient()
# Test GET request