This document provides a comprehensive, detailed, and professional output for the "Generate Code" step of your API Integration Builder workflow. The goal is to deliver clean, well-commented, and production-ready code that facilitates seamless integration with an external RESTful API.
For this deliverable, we have chosen Python as the programming language due to its versatility, rich ecosystem of libraries, and readability, making it an excellent choice for API integrations. We will demonstrate integration with a conceptual "Product Catalog API" to illustrate common API interaction patterns such as fetching data (GET) and creating new resources (POST).
This output provides a modular and robust Python codebase designed to interact with a typical RESTful API. It emphasizes best practices in API client design, error handling, configuration management, and readability.
The generated code will allow you to:
Before diving into the code, it's important to understand the fundamental concepts guiding this integration:
requests library in Python is used to send HTTP requests to the API.API Key authentication method is implemented, where the key is sent in a custom HTTP header.The generated solution consists of three main Python files and a requirements.txt file:
config.py: Stores environment-specific configurations like the API base URL and API key.api_client.py: Contains the core logic for interacting with the external API. This file defines a class (ProductCatalogAPIClient) that encapsulates all API calls, authentication, and basic error handling.main.py: An example script demonstrating how to use the ProductCatalogAPIClient to perform various operations and print the results.requirements.txt: Lists all necessary Python packages required to run the integration.Conceptual Product Catalog API Endpoints:
GET /products: Retrieve a list of products.GET /products/{id}: Retrieve a single product by ID.POST /products: Create a new product.Below is the generated code, structured into its respective files.
requirements.txt--- #### `api_client.py`
python
import requests
import logging
from typing import Dict, Any, List, Optional
from config import API_BASE_URL, API_KEY, LOGGING_LEVEL, LOG_FILE_PATH
logger = logging.getLogger(__name__)
logger.setLevel(LOGGING_LEVEL)
ch = logging.StreamHandler()
ch.setLevel(LOGGING_LEVEL)
fh = logging.FileHandler(LOG_FILE_PATH)
fh.setLevel(LOGGING_LEVEL)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
ch.setFormatter(formatter)
fh.setFormatter(formatter)
if not logger.handlers: # Prevent adding handlers multiple times if module is reloaded
logger.addHandler(ch)
logger.addHandler(fh)
class APIIntegrationError(Exception):
"""Custom exception for API integration-specific errors."""
def __init__(self, message: str, status_code: Optional[int] = None, details: Any = None):
super().__init__(message)
self.status_code = status_code
self.details = details
class ProductCatalogAPIClient:
"""
A client for interacting with the Product Catalog API.
Encapsulates all API calls, authentication, and error handling.
"""
def __init__(self, base_url: str, api_key: str):
"""
Initializes the API client with the base URL and API key.
Args:
base_url (str): The base URL of the Product Catalog 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:
raise ValueError("API Key cannot be empty.")
self.base_url = base_url.rstrip('/') # Ensure no trailing slash for consistent URL building
self.api_key = api_key
self.session = requests.Session() # Use a session for connection pooling and persistent headers
self._setup_session_headers()
logger.info(f"ProductCatalogAPIClient initialized for base URL: {self.base_url}")
def _setup_session_headers(self):
"""Sets up common headers for the requests session."""
self.session.headers.update({
"Content-Type": "application/json",
"Accept": "application/json",
"X-API-Key": self.api_key, # Custom header for API Key authentication
"User-Agent": "ProductCatalogIntegration/1.0.0 (Python)"
})
logger.debug("Session headers configured.")
def _make_request(self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None, json_data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
Internal helper method to make an HTTP request and handle common responses.
Args:
method (str): The HTTP method (e.g., "GET", "POST").
endpoint (str): The API endpoint (e.g., "/products").
params (Optional[Dict[str, Any]]): Dictionary of URL query parameters.
json_data (Optional[Dict[str, Any]]): Dictionary of JSON data to send in the request body.
Returns:
Dict[str, Any]: The JSON response from the API.
Raises:
APIIntegrationError: If the request fails or returns an error status code.
"""
url = f"{self.base_url}{endpoint}"
logger.debug(f"Making {method} request to: {url} with params: {params}, data: {json_data}")
try:
response = self.session.request(method, url, params=params, json=json_data, timeout=10)
response.raise_for_status() # Raises HTTPError for bad responses (4xx or 5xx)
logger.debug(f"Received successful response (Status: {response.status_code}) from {url}")
# Attempt to parse JSON response. Some successful responses might have no body.
if response.text:
return response.json()
return {} # Return empty dict for responses with no content (e.g., 204 No Content)
except requests.exceptions.HTTPError as e:
status_code = e.response.status_code
error_details = None
try:
error_details = e.response.json()
except requests.exceptions.JSONDecodeError:
error_details = e.response.text # Fallback to raw text if not JSON
logger.error(f"HTTP Error {status_code} for {method} {url}: {error_details}")
raise APIIntegrationError(
f"API request failed with status {status_code}",
status_code=status_code,
details=error_details
) from e
except requests.exceptions.ConnectionError as e:
logger.error(f"Connection Error for {method} {url}: {e}")
raise APIIntegrationError(f"Failed to connect to the API: {e}", details=str(e)) from e
except requests.exceptions.Timeout as e:
logger.error(f"Timeout Error for {method} {url}: {e}")
raise APIIntegrationError(f"API request timed out: {e}", details=str(e)) from e
except requests.exceptions.RequestException as e:
logger.error(f"An unexpected Request Error occurred for {method} {url}: {e}")
raise APIIntegrationError(f"An unexpected request error occurred: {e}", details=str(e)) from e
except Exception as e:
logger.critical(f"An unhandled error occurred during API request for {method} {url}: {e}")
raise APIIntegrationError(f"An unknown error occurred: {e}", details=str(e)) from e
def get_products(self, limit: int = 10, skip: int = 0) -> List[Dict[str, Any]]:
"""
Retrieves a list of products from the API.
Args:
limit (int): The maximum number of products to return (default: 10).
skip (int): The number of products to skip for pagination (default: 0).
Returns:
List[Dict[str, Any]]: A list of product dictionaries.
"""
endpoint = "/products"
params = {"limit": limit, "skip": skip}
logger.info(f"Fetching products with limit={limit}, skip={skip}")
response_data = self._make_request("GET", endpoint, params=params)
return response_data.get("products", []) # Assuming the API returns a 'products' key
def get_product_by_id(self, product_id: str) -> Dict[str, Any]:
"""
Retrieves a single product by its ID.
Args:
product_id (str): The unique identifier of the product.
Returns:
Dict[str, Any]: The product dictionary.
Raises:
APIIntegrationError: If the product is not found or other API error occurs.
"""
if not product_id:
raise ValueError("Product ID cannot be empty.")
endpoint = f"/products/{product_id}"
logger.info(f"Fetching product with ID: {product_id}")
return self._make_request("GET", endpoint)
def create_product(self, product_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Creates a new product in the API.
Args:
product_data (Dict[str, Any]): A dictionary containing the product details.
Example: {"name": "New Widget", "price": 29.99, "category": "Electronics"}
Returns:
Dict[str, Any]: The newly created product object, including its ID.
"""
if not product_data:
raise ValueError("Product data cannot be empty for creation.")
endpoint = "/products"
logger.info(f"Creating new product with data: {product_data}")
return self._make_request("POST", endpoint, json_data=product_data)
# You can add more methods here for other API operations (e.g., update_product, delete_product)
# def update_product(self, product_id: str, product_data: Dict[str, Any]) -> Dict[str, Any]:
# endpoint = f"/products/{product_id}"
# return self._make_request("PUT", endpoint, json_data=product_data)
# def delete_product(self, product_id: str) -> None:
# endpoint = f"/products/{product_id}"
# self._make_
projectmanager → create_project)Workflow Step: 2 of 2
This deliverable provides a comprehensive guide and code examples for integrating with an external Project Management System (PMS) API to programmatically create new projects. This is a critical step in automating project initiation workflows, ensuring consistency and reducing manual effort.
The goal of this integration is to enable your system to interact with a third-party Project Management System (e.g., Jira, Asana, Trello, Monday.com, custom internal PMS) and create a new project entry through its API. This involves sending a structured request with project details and handling the system's response.
Key Objectives:
create_projectTo successfully create a project via an external API, the following components are typically required:
POST * https://api.externalpms.com/v1/projects
* https://your-pms-instance.com/api/projects
Example:* For a hypothetical PMS, it might be https://api.projectmanagerpro.com/v2/workspaces/{workspace_id}/projects
The POST request will typically carry a JSON payload containing all the necessary details for the new project. Common fields include:
name (string, required): The title or name of the project.description (string, optional): A detailed description of the project.startDate (string, optional): The planned start date of the project (e.g., "YYYY-MM-DD").endDate (string, optional): The planned end date of the project (e.g., "YYYY-MM-DD").ownerId / managerId (string, optional): The ID of the user responsible for the project.status (string, optional): Initial status of the project (e.g., "Active", "Planning", "On Hold").teamMembers (array of strings, optional): IDs of initial team members.customFields (object, optional): Any additional custom fields defined in the PMS.Example JSON Request Body:
{
"name": "PantheraHive Q3 Marketing Campaign",
"description": "Develop and launch a comprehensive marketing campaign for Q3 product releases.",
"startDate": "2023-07-01",
"endDate": "2023-09-30",
"ownerId": "usr_abc123def456",
"status": "Planning",
"teamMembers": [
"usr_abc123def456",
"usr_ghi789jkl012"
],
"customFields": {
"budget_allocated": 75000,
"priority": "High"
}
}
Secure access to the API is paramount. Common authentication methods include:
Header Example:* Authorization: Api-Key YOUR_API_KEY
Authorization header. Header Example:* Authorization: Bearer YOUR_ACCESS_TOKEN
Authorization header.Recommendation: Use environment variables or a secure configuration management system to store API keys and tokens, never hardcode them directly into your application.
After sending the request, your system must handle the API's response:
* The API will typically return a JSON object containing the newly created project's details, including its unique ID (projectId).
Example Response:*
{
"id": "proj_xyz789abc012",
"name": "PantheraHive Q3 Marketing Campaign",
"description": "...",
"status": "Planning",
"createdAt": "2023-06-15T10:00:00Z"
}
* 400 Bad Request: Invalid request payload (e.g., missing required fields, incorrect data types).
* 401 Unauthorized: Missing or invalid authentication credentials.
* 403 Forbidden: Authenticated but not authorized to perform the action.
* 404 Not Found: The endpoint or a referenced resource (e.g., workspace_id) does not exist.
* 429 Too Many Requests: Rate limit exceeded.
* 500 Internal Server Error: An unexpected error on the PMS server.
Error responses usually include a JSON body with an error message and/or code for debugging.*
This example demonstrates how to integrate with a hypothetical Project Management System API using Python's requests library.
requests library: Install via pip install requestsStore your API details securely. For this example, we'll use placeholder variables.
import requests
import os
import json
# --- Configuration ---
# Base URL for the external Project Management System API
PMS_API_BASE_URL = os.getenv("PMS_API_BASE_URL", "https://api.projectmanagerpro.com/v2")
# Your API Key or Bearer Token (securely retrieved from environment variables)
# Example for API Key:
PMS_API_KEY = os.getenv("PMS_API_KEY", "YOUR_SECURE_API_KEY_HERE")
# Example for Bearer Token:
# PMS_BEARER_TOKEN = os.getenv("PMS_BEARER_TOKEN", "YOUR_SECURE_BEARER_TOKEN_HERE")
# Identifier for the workspace/organization where projects should be created
# This might be needed in the URL or payload, depending on the API.
WORKSPACE_ID = os.getenv("PMS_WORKSPACE_ID", "your_workspace_id_example")
def create_project(project_data: dict) -> dict:
"""
Creates a new project in the external Project Management System.
Args:
project_data (dict): A dictionary containing the project's details.
Example: {
"name": "New Project",
"description": "...",
"startDate": "YYYY-MM-DD",
"ownerId": "usr_id"
}
Returns:
dict: A dictionary containing the newly created project's ID and details
on success, or an error dictionary on failure.
"""
endpoint = f"{PMS_API_BASE_URL}/workspaces/{WORKSPACE_ID}/projects"
headers = {
"Content-Type": "application/json",
# Use either API Key or Bearer Token based on your PMS API
"Authorization": f"Api-Key {PMS_API_KEY}" # For API Key authentication
# "Authorization": f"Bearer {PMS_BEARER_TOKEN}" # For Bearer Token authentication
}
try:
print(f"Attempting to create project: {project_data.get('name')}...")
response = requests.post(endpoint, headers=headers, json=project_data, timeout=10)
response.raise_for_status() # Raise an HTTPError for bad responses (4xx or 5xx)
response_json = response.json()
print(f"Project '{project_data.get('name')}' created successfully!")
print(f"New Project ID: {response_json.get('id')}")
return {
"status": "success",
"project_id": response_json.get('id'),
"details": response_json
}
except requests.exceptions.HTTPError as http_err:
print(f"HTTP error occurred: {http_err}")
print(f"Response Status Code: {response.status_code}")
try:
error_details = response.json()
print(f"Error Details: {json.dumps(error_details, indent=2)}")
return {
"status": "error",
"message": f"API responded with status {response.status_code}: {error_details.get('message', 'No specific message provided.')}",
"details": error_details,
"status_code": response.status_code
}
except json.JSONDecodeError:
print(f"Could not parse error response as JSON: {response.text}")
return {
"status": "error",
"message": f"API responded with status {response.status_code}. Raw response: {response.text}",
"status_code": response.status_code
}
except requests.exceptions.ConnectionError as conn_err:
print(f"Connection error occurred: {conn_err}")
return {
"status": "error",
"message": f"Network connection error: {conn_err}"
}
except requests.exceptions.Timeout as timeout_err:
print(f"Request timed out: {timeout_err}")
return {
"status": "error",
"message": f"Request to PMS API timed out: {timeout_err}"
}
except requests.exceptions.RequestException as req_err:
print(f"An unexpected request error occurred: {req_err}")
return {
"status": "error",
"message": f"An unexpected error occurred during the API request: {req_err}"
}
except Exception as e:
print(f"An unexpected error occurred: {e}")
return {
"status": "error",
"message": f"An unforeseen error occurred: {e}"
}
if __name__ == "__main__":
# Example project data
new_project_data = {
"name": "PantheraHive AI Integration Phase 1",
"description": "Initiate the first phase of AI integration across core PantheraHive services.",
"startDate": "2023-08-01",
"endDate": "2023-12-31",
"ownerId": "