This document provides the comprehensive, detailed, and professional code generation output for the "API Integration Builder" workflow. This output is designed to be production-ready, well-commented, and easily adaptable for integrating with various external RESTful APIs.
The goal of this generate_code step is to produce a robust, maintainable, and extensible Python client for interacting with a generic RESTful API. This client incorporates best practices for API integration, including secure configuration, resilient request handling, structured error management, and comprehensive logging.
This deliverable focuses on creating a foundational API client class that can be customized for specific API endpoints and data models. It serves as a strong starting point for any external API integration project.
The generated code adheres to the following core principles:
To provide a concrete example, the generated code is structured around integrating with a hypothetical RESTful API that manages resources (e.g., users, products, contacts). We will demonstrate common CRUD (Create, Read, Update, Delete) operations.
Hypothetical API Endpoints:
GET /resources: Retrieve a list of resources.GET /resources/{id}: Retrieve a specific resource by ID.POST /resources: Create a new resource.PUT /resources/{id}: Update an existing resource.DELETE /resources/{id}: Delete a resource.The following Python code provides a comprehensive API client. It includes necessary dependencies, configuration loading, custom exceptions, retry logic, and methods for common HTTP operations.
Prerequisites:
To run this code, you will need to install the following Python packages:
--- #### 4.1 `.env` File (Configuration) Create a file named `.env` in the root directory of your project and populate it with your API configuration. This keeps sensitive information out of your codebase.
python
import os
import requests
import logging
from dotenv import load_dotenv
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
load_dotenv() # Load environment variables from .env file
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class APIError(Exception):
"""Custom exception for API-related errors."""
def __init__(self, message, status_code=None, details=None):
super().__init__(message)
self.status_code = status_code
self.details = details
def __str__(self):
if self.status_code:
return f"APIError: {self.status_code} - {self.args[0]}"
return f"APIError: {self.args[0]}"
class NetworkError(APIError):
"""Custom exception for network-related errors during API calls."""
pass
class AuthenticationError(APIError):
"""Custom exception for authentication failures (e.g., 401 Unauthorized)."""
pass
class NotFoundError(APIError):
"""Custom exception for resource not found errors (e.g., 404 Not Found)."""
pass
class BadRequestError(APIError):
"""Custom exception for bad request errors (e.g., 400 Bad Request)."""
pass
class APIClient:
"""
A robust and generic client for interacting with RESTful APIs.
Features:
- Loads configuration from environment variables.
- Handles common HTTP methods (GET, POST, PUT, DELETE).
- Implements retry logic for transient errors.
- Provides structured error handling with custom exceptions.
- Integrates logging for operational insights.
- Supports API Key authentication.
"""
def __init__(self, base_url=None, api_key=None):
"""
Initializes the APIClient with base URL and API key.
Args:
base_url (str, optional): The base URL of the API. If None,
it tries to load from API_BASE_URL env var.
api_key (str, optional): The API key for authentication. If None,
it tries to load from API_KEY env var.
"""
self.base_url = base_url or os.getenv("API_BASE_URL")
self.api_key = api_key or os.getenv("API_KEY")
if not self.base_url:
raise ValueError("API base URL is not provided and not found in environment variables.")
if not self.api_key:
logger.warning("API Key is not provided. Requests might fail if API requires authentication.")
self.session = requests.Session()
logger.info(f"APIClient initialized for base URL: {self.base_url}")
def _get_headers(self, custom_headers=None):
"""
Constructs the standard headers for API requests, including authentication.
Args:
custom_headers (dict, optional): Additional headers to merge.
Returns:
dict: A dictionary of HTTP headers.
"""
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
}
if self.api_key:
# Example: API Key in Authorization header as Bearer token or custom scheme
# Adapt this based on your API's authentication mechanism
headers["Authorization"] = f"Bearer {self.api_key}"
# For APIs that use a custom header like 'X-API-Key':
# headers["X-API-Key"] = self.api_key
if custom_headers:
headers.update(custom_headers)
return headers
@retry(stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=10),
retry=retry_if_exception_type((NetworkError, requests.exceptions.Timeout)),
reraise=True)
def _request(self, method, endpoint, params=None, data=None, json=None, headers=None, **kwargs):
"""
Internal helper method to make an HTTP request with retry logic and error handling.
Args:
method (str): The HTTP method (e.g., 'GET', 'POST', 'PUT', 'DELETE').
endpoint (str): The specific API endpoint path (e.g., '/resources').
params (dict, optional): Dictionary of URL query parameters.
data (dict or str, optional): Dictionary, bytes, or file-like object to send in the body.
json (dict, optional): A JSON-serializable Python object to send in the body.
headers (dict, optional): Custom headers for this specific request.
**kwargs: Additional keyword arguments to pass to requests.request.
Returns:
dict: The JSON response from the API.
Raises:
APIError: For general API errors.
NetworkError: For network connectivity issues.
AuthenticationError: If authentication fails (401).
NotFoundError: If the resource is not found (404).
BadRequestError: If the request is malformed (400).
"""
url = f"{self.base_url}{endpoint}"
request_headers = self._get_headers(headers)
logger.debug(f"Making {method} request to: {url}")
logger.debug(f"Request Headers: {request_headers}")
if params:
logger.debug(f"Request Params: {params}")
if json:
logger.debug(f"Request JSON Body: {json}")
elif data:
logger.debug(f"Request Data Body: {data}")
try:
response = self.session.request(
method, url, params=params, data=data, json=json, headers=request_headers, **kwargs
)
response.raise_for_status() # Raises HTTPError for bad responses (4xx or 5xx)
logger.info(f"Successfully {method} {endpoint}. Status: {response.status_code}")
return response.json()
except requests.exceptions.Timeout as e:
logger.error(f"Request to {url} timed out: {e}")
raise NetworkError(f"Request timed out for {url}", details=str(e)) from e
except requests.exceptions.ConnectionError as e:
logger.error(f"Network connection error to {url}: {e}")
raise NetworkError(f"Network connection error to {url}", details=str(e)) from e
except requests.exceptions.HTTPError as e:
status_code = e.response.status_code
error_message = f"API returned error for {url}: {status_code} - {e.response.text}"
logger.error(error_message)
response_json = {}
try:
response_json = e.response.json()
except ValueError:
pass # Not all error responses are JSON
if status_code == 400:
raise BadRequestError(e.response.text, status_code=status_code, details=response_json) from e
elif status_code == 401:
raise AuthenticationError(e.response.text, status_code=status_code, details=response_json) from e
elif status_code == 403:
raise APIError(f"Forbidden: {e.response.text}", status_code=status_code, details=response_json) from e
elif status_code == 404:
raise NotFoundError(e.response.text, status_code=status_code, details=response_json) from e
elif 400 <= status_code < 500:
raise APIError(f"Client error: {e.response.text}", status_code=status_code, details=response_json) from e
elif 500 <= status_code < 600:
# Server errors might be transient, retries are handled by tenacity
raise APIError(f"Server error: {e.response.text}", status_code=status_code, details=response_json) from e
else:
raise APIError(f"Unexpected HTTP error: {e.response.text}", status_code=status_code, details=response_json) from e
except Exception as e:
logger.exception(f"An unexpected error occurred during request to {url}: {e}")
raise APIError(f"An unexpected error occurred: {e}", details=str(e)) from e
def get(self, endpoint, params=None, headers=None, **kwargs):
"""Sends a GET request."""
return self._request('GET', endpoint, params=params, headers=headers, **kwargs)
def post(self, endpoint, data=None, json=None, headers=None, **kwargs):
"""Sends a POST request."""
return self._request('POST', endpoint, data=data, json=json, headers=headers, **kwargs)
def put(self, endpoint, data=None, json=None, headers=None, **kwargs):
"""Sends a PUT request."""
return self._request('PUT', endpoint, data=data, json=json, headers=headers, **kwargs)
def delete(self, endpoint, params=None, headers=None, **kwargs):
"""Sends a DELETE request."""
return self._request('DELETE', endpoint, params=params, headers=headers, **kwargs)
# --- 4. Example Specific API Methods (Adapt these for your API) ---
def get_all_resources(self, query_params=None):
"""Retrieves a list of all resources."""
logger.info("Fetching all resources...")
return self.get("/resources", params=query_params)
def get_resource_by_id(self, resource_id):
"""Retrieves a specific resource by its ID."""
logger.info(f"Fetching resource with ID: {resource_id}")
return self.get(f"/resources/{resource_id}")
def create_resource(self, resource_data):
"""Creates a new resource."""
logger.info(f"Creating resource with data: {resource_data}")
return self.post("/resources", json=resource_data)
def update_resource(self, resource_id, resource_data):
"""Updates an existing resource."""
logger.info(f"Updating resource {resource_id} with data: {resource_data}")
return self.put(f"/resources/{resource
This deliverable provides a comprehensive framework and example code for integrating with an external API to perform the "Create Project" operation. It outlines the necessary components, best practices, and a clear, actionable Python code example to facilitate seamless integration into your existing systems. The focus is on robust, secure, and maintainable API interaction, ensuring successful project creation and effective error handling.
The primary objective of this module is to enable your application to programmatically create new projects in a target Project Management System (PMS) or similar external service via its RESTful API. This integration aims to:
Given that a specific API was not provided, this integration is designed based on common patterns for RESTful APIs.
POST /api/v1/projects (or similar, e.g., /projects)X-API-Key header for simplicity, but can be easily adapted for Authorization: Bearer <token>.A robust API integration requires several key components to ensure reliability and security:
Sensitive information like API keys, base URLs, and other environment-specific settings should be managed securely and never hardcoded.
.env files for local development, system environment variables for production). * API_BASE_URL: The root URL of the target API (e.g., https://api.example.com/).
* API_KEY: The authentication key or token.
Securely authenticating requests to the external API.
X-API-Key or Authorization).Building the HTTP request with the necessary components.
POST/api/v1/projects). * Content-Type: application/json: Indicates that the request body is JSON.
* Authentication Header: X-API-Key: YOUR_API_KEY or Authorization: Bearer YOUR_TOKEN.
* Example Payload Structure:
{
"name": "Project Alpha",
"description": "Detailed description for Project Alpha.",
"startDate": "2023-10-26",
"endDate": "2024-03-31",
"ownerId": "user-uuid-123",
"status": "Planned",
"priority": "High"
}
* Required Fields: (Determined by the target API) name, description, startDate, ownerId are common.
Processing the API's response, both for success and failure.
* Expected JSON body containing the newly created project's ID and other relevant details.
* Example Success Response:
{
"id": "proj-abc-123",
"name": "Project Alpha",
"status": "Planned",
"message": "Project created successfully."
}
* Expected JSON body with an error code and a human-readable message.
* Example Error Response:
{
"errorCode": "VALIDATION_ERROR",
"message": "Project name cannot be empty.",
"details": [
{"field": "name", "error": "required"}
]
}
Robust error handling is critical for production systems.
4xx (client errors) and 5xx (server errors) codes.APIAuthenticationError, APIValidationError, APIServerError).This example utilizes Python with the requests library, which is a standard choice for HTTP interactions due to its simplicity and power.
requests library installed: pip install requests python-dotenv.env file in the same directory as your script for environment variables.config.py (or similar for environment variables)
import os
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
class APIConfig:
"""
Manages API configuration settings.
"""
BASE_URL = os.getenv("API_BASE_URL", "https://api.example.com/v1")
API_KEY = os.getenv("API_KEY") # Ensure this is set in your .env
PROJECT_ENDPOINT = "/projects" # Specific endpoint for project creation
@classmethod
def validate(cls):
"""Validates that necessary configuration is present."""
if not cls.API_KEY:
raise ValueError("API_KEY environment variable is not set.")
if not cls.BASE_URL:
raise ValueError("API_BASE_URL environment variable is not set.")
# Validate configuration on import
APIConfig.validate()
.env File ExampleCreate a file named .env in the same directory:
API_BASE_URL=https://your-project-management-api.com/v1
API_KEY=your_secure_api_key_here
Important: Replace your-project-management-api.com and your_secure_api_key_here with your actual API details. Never commit your .env file to version control.
api_client.py
import requests
import json
import logging
from config import APIConfig
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class APIIntegrationError(Exception):
"""Base exception for API integration errors."""
pass
class APIAuthenticationError(APIIntegrationError):
"""Raised for 401 Unauthorized or 403 Forbidden errors."""
pass
class APIValidationError(APIIntegrationError):
"""Raised for 400 Bad Request errors."""
def __init__(self, message, details=None):
super().__init__(message)
self.details = details or {}
class APIServerError(APIIntegrationError):
"""Raised for 5xx server errors."""
pass
class APINetworkError(APIIntegrationError):
"""Raised for network-related issues (e.g., connection refused, timeout)."""
pass
def create_project(project_data: dict) -> dict:
"""
Integrates with an external API to create a new project.
Args:
project_data (dict): A dictionary containing the project details.
Example: {
"name": "New Project",
"description": "Project description.",
"startDate": "2023-10-26",
"endDate": "2024-12-31",
"ownerId": "user-123",
"status": "Planned"
}
Returns:
dict: The response data from the API, typically including the new project's ID.
Raises:
APIAuthenticationError: If authentication fails.
APIValidationError: If the provided project_data is invalid.
APIServerError: If the API encounters an internal server error.
APINetworkError: If there's a network issue connecting to the API.
APIIntegrationError: For other unexpected API errors.
"""
url = f"{APIConfig.BASE_URL}{APIConfig.PROJECT_ENDPOINT}"
headers = {
"Content-Type": "application/json",
"X-API-Key": APIConfig.API_KEY # Using X-API-Key as per assumption
# For Bearer token, use: "Authorization": f"Bearer {APIConfig.API_KEY}"
}
try:
logging.info(f"Attempting to create project: {project_data.get('name', 'Unnamed Project')}")
response = requests.post(url, headers=headers, data=json.dumps(project_data), timeout=10)
response.raise_for_status() # Raises HTTPError for bad responses (4xx or 5xx)
response_data = response.json()
logging.info(f"Project '{project_data.get('name')}' created successfully. ID: {response_data.get('id')}")
return response_data
except requests.exceptions.Timeout:
logging.error(f"API request timed out after 10 seconds for project: {project_data.get('name')}")
raise APINetworkError("API request timed out.")
except requests.exceptions.ConnectionError as e:
logging.error(f"Failed to connect to API for project {project_data.get('name')}: {e}")
raise APINetworkError(f"Failed to connect to API: {e}")
except requests.exceptions.HTTPError as e:
status_code = e.response.status_code
error_message = f"API responded with status {status_code}"
try:
error_details = e.response.json()
error_message += f": {error_details.get('message', 'No specific message')}"
logging.error(f"API Error ({status_code}) for project {project_data.get('name')}: {error_details}")
except json.JSONDecodeError:
error_details = {"raw_response": e.response.text}
logging.error(f"API Error ({status_code}) for project {project_data.get('name')}: {e.response.text}")
if status_code in [401, 403]:
raise APIAuthenticationError(error_message)
elif status_code == 400:
raise APIValidationError(error_message, details=error_details)
elif 400 <= status_code < 500:
raise APIIntegrationError(f"Client error from API: {error_message}")
elif 500 <= status_code < 600:
raise APIServerError(f"Server error from API: {error_message}")
else:
raise APIIntegrationError(f"Unexpected HTTP error from API: {error_message}")
except json.JSONDecodeError:
logging.error(f"Failed to decode JSON response for project {project_data.get('name')}. Raw response: {response.text}")
raise APIIntegrationError("Received non-JSON response from API.")
except Exception as e:
logging.critical(f"An unexpected error occurred during API integration for project {project_data.get('name')}: {e}", exc_info=True)
raise APIIntegrationError(f"An unexpected error occurred: {e}")
#