API Integration Builder: Code Generation Deliverable (Step 1 of 2)

Project: API Integration Builder

Step: collab → generate_code

Description: Generate code to integrate with external APIs

1. Introduction

This document provides a comprehensive, detailed, and professional code generation output for the "API Integration Builder" workflow. This is Step 1 of 2, focusing on generating the core integration code. The generated code is designed to be clean, well-commented, and production-ready, facilitating seamless integration with external RESTful APIs.

For demonstration purposes, we will use a common pattern for interacting with a generic REST API, specifically illustrating operations against the widely-used JSONPlaceholder API, which provides fake online REST API for testing and prototyping. This example will cover basic CRUD (Create, Read, Update, Delete) operations, error handling, and best practices applicable to most RESTful integrations.

2. Core API Integration Code (Python)

The following Python code provides a robust framework for interacting with a generic RESTful API. It utilizes the requests library, a standard for HTTP requests in Python, and incorporates essential features like proper error handling, structured data processing, and extensibility.

import requests
import json
import os
from typing import Dict, Any, List, Optional

# --- Configuration ---
# It's best practice to manage sensitive information and base URLs via environment variables
# or a secure configuration management system.
# For demonstration, we'll set a default, but recommend using os.getenv() in production.
DEFAULT_API_BASE_URL = os.getenv("API_BASE_URL", "https://jsonplaceholder.typicode.com")
DEFAULT_API_KEY = os.getenv("API_KEY", None) # Example for APIs requiring a key

class APIIntegrationError(Exception):
    """Custom exception for API integration specific errors."""
    pass

class APIClient:
    """
    A client for interacting with a generic RESTful API.

    Attributes:
        base_url (str): The base URL of the API (e.g., "https://api.example.com/v1").
        headers (Dict[str, str]): Default headers to be sent with each request,
                                   including authentication if applicable.
        timeout (int): Default timeout for HTTP requests in seconds.
    """

    def __init__(self, base_url: str = DEFAULT_API_BASE_URL, api_key: Optional[str] = DEFAULT_API_KEY, timeout: int = 30):
        """
        Initializes the APIClient with the base URL and optional API key.

        Args:
            base_url (str): The base URL for the API.
            api_key (Optional[str]): An optional API key for authentication.
                                      If provided, it will be added to headers.
            timeout (int): The default timeout for requests in seconds.
        """
        if not base_url:
            raise ValueError("API base URL cannot be empty.")

        self.base_url = base_url
        self.timeout = timeout
        self.headers: Dict[str, str] = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            # Add other common headers here, e.g., User-Agent
        }

        if api_key:
            # Example: For APIs using 'X-API-Key' header. Adjust as per your API's auth method.
            # Common methods: 'Authorization: Bearer <token>', 'X-API-Key: <key>'
            print(f"DEBUG: Initializing APIClient with API Key: {'*' * len(api_key) if api_key else 'None'}")
            self.headers["X-API-Key"] = api_key
            # For JSONPlaceholder, no API key is needed, this is just an example for other APIs.

    def _request(self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None,
                 params: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None) -> Any:
        """
        Internal helper method to make an HTTP request.

        Args:
            method (str): The HTTP method (e.g., 'GET', 'POST', 'PUT', 'DELETE').
            endpoint (str): The specific API endpoint (e.g., '/posts', '/users/1').
            data (Optional[Dict[str, Any]]): Dictionary of data to send in the request body (for POST/PUT).
            params (Optional[Dict[str, Any]]): Dictionary of query parameters to send with the request.
            headers (Optional[Dict[str, str]]): Additional headers to merge with default headers.

        Returns:
            Any: The JSON response from the API.

        Raises:
            APIIntegrationError: If the API request fails or returns an error status.
        """
        url = f"{self.base_url}{endpoint}"
        _headers = {**self.headers, **(headers or {})} # Merge default and custom headers

        try:
            print(f"DEBUG: Making {method} request to {url} with params={params}, data={data}, headers={_headers}")
            if method.upper() == "GET":
                response = requests.get(url, params=params, headers=_headers, timeout=self.timeout)
            elif method.upper() == "POST":
                response = requests.post(url, json=data, params=params, headers=_headers, timeout=self.timeout)
            elif method.upper() == "PUT":
                response = requests.put(url, json=data, params=params, headers=_headers, timeout=self.timeout)
            elif method.upper() == "DELETE":
                response = requests.delete(url, params=params, headers=_headers, timeout=self.timeout)
            else:
                raise ValueError(f"Unsupported HTTP method: {method}")

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

            # Attempt to parse JSON, handle cases where response might be empty (e.g., 204 No Content)
            if response.status_code == 204:
                return None # No content to return
            return response.json()

        except requests.exceptions.HTTPError as http_err:
            error_details = response.text
            status_code = response.status_code
            print(f"ERROR: HTTP error occurred: {http_err} - Status: {status_code}, Details: {error_details}")
            raise APIIntegrationError(
                f"API request failed with status {status_code}: {error_details}"
            ) from http_err
        except requests.exceptions.ConnectionError as conn_err:
            print(f"ERROR: Connection error occurred: {conn_err}")
            raise APIIntegrationError(
                f"Could not connect to the API at {self.base_url}: {conn_err}"
            ) from conn_err
        except requests.exceptions.Timeout as timeout_err:
            print(f"ERROR: Request timed out: {timeout_err}")
            raise APIIntegrationError(
                f"API request timed out after {self.timeout} seconds: {timeout_err}"
            ) from timeout_err
        except requests.exceptions.RequestException as req_err:
            print(f"ERROR: An unexpected request error occurred: {req_err}")
            raise APIIntegrationError(
                f"An unexpected error occurred during the API request: {req_err}"
            ) from req_err
        except json.JSONDecodeError as json_err:
            print(f"ERROR: Failed to decode JSON response: {json_err} - Response: {response.text}")
            raise APIIntegrationError(
                f"Failed to decode JSON response from {url}: {json_err}"
            ) from json_err

    def get_resource(self, endpoint: str, resource_id: Optional[Any] = None,
                     params: Optional[Dict[str, Any]] = None) -> Any:
        """
        Fetches a single resource or a list of resources from the API.

        Args:
            endpoint (str): The base endpoint (e.g., '/posts').
            resource_id (Optional[Any]): The ID of a specific resource to fetch (e.g., 1 for '/posts/1').
            params (Optional[Dict[str, Any]]): Query parameters for filtering/pagination.

        Returns:
            Any: The fetched resource (dict) or list of resources (list).
        """
        full_endpoint = f"{endpoint}/{resource_id}" if resource_id is not None else endpoint
        print(f"DEBUG: Getting resource from {full_endpoint}")
        return self._request("GET", full_endpoint, params=params)

    def create_resource(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Creates a new resource on the API.

        Args:
            endpoint (str): The endpoint to create the resource (e.g., '/posts').
            data (Dict[str, Any]): The data for the new resource.

        Returns:
            Dict[str, Any]: The newly created resource as returned by the API.
        """
        print(f"DEBUG: Creating resource at {endpoint} with data: {data}")
        return self._request("POST", endpoint, data=data)

    def update_resource(self, endpoint: str, resource_id: Any, data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Updates an existing resource on the API.

        Args:
            endpoint (str): The base endpoint (e.g., '/posts').
            resource_id (Any): The ID of the resource to update.
            data (Dict[str, Any]): The data to update the resource with.

        Returns:
            Dict[str, Any]: The updated resource as returned by the API.
        """
        full_endpoint = f"{endpoint}/{resource_id}"
        print(f"DEBUG: Updating resource at {full_endpoint} with data: {data}")
        return self._request("PUT", full_endpoint, data=data)

    def delete_resource(self, endpoint: str, resource_id: Any) -> Optional[None]:
        """
        Deletes a resource from the API.

        Args:
            endpoint (str): The base endpoint (e.g., '/posts').
            resource_id (Any): The ID of the resource to delete.

        Returns:
            Optional[None]: None if the deletion was successful (often 204 No Content).
        """
        full_endpoint = f"{endpoint}/{resource_id}"
        print(f"DEBUG: Deleting resource at {full_endpoint}")
        return self._request("DELETE", full_endpoint)

# --- Example Usage (for demonstration purposes) ---
if __name__ == "__main__":
    print("--- API Integration Builder: Example Usage ---")

    # Initialize the client (using JSONPlaceholder by default)
    # In a real application, you might pass your specific base_url and api_key here.
    # client = APIClient(base_url="https://my-custom-api.com/v1", api_key="YOUR_API_KEY")
    client = APIClient()

    print("\n--- 1. Fetching a list of resources (e.g., all posts) ---")
    try:
        posts = client.get_resource("/posts", params={"_limit": 5}) # Get first 5 posts
        if posts:
            print(f"Successfully fetched {len(posts)} posts. First post title: '{posts[0].get('title')}'")
        else:
            print("No posts fetched.")
    except APIIntegrationError as e:
        print(f"Error fetching posts: {e}")

    print("\n--- 2. Fetching a single resource (e.g., post with ID 1) ---")
    try:
        post_id = 1
        single_post = client.get_resource("/posts", resource_id=post_id)
        if single_post:
            print(f"Successfully fetched post {post_id}: '{single_post.get('title')}'")
        else:
            print(f"Post {post_id} not found.")
    except APIIntegrationError as e:
        print(f"Error fetching post {post_id}: {e}")

    print("\n--- 3. Creating a new resource (e.g., a new post) ---")
    new_post_data = {
        "title": "My New Integration Post",
        "body": "This is the content of my new post, created via API integration.",
        "userId": 1
    }
    try:
        created_post = client.create_resource("/posts", data=new_post_data)
        print(f"Successfully created post with ID: {created_post.get('id')}, Title: '{created_post.get('title')}'")
    except APIIntegrationError as e:
        print(f"Error creating post: {e}")

    print("\n--- 4. Updating an existing resource (e.g., the newly created post) ---")
    if 'created_post' in locals() and created_post and created_post.get('id'):
        post_to_update_id = created_post['id']
        updated_post_data = {
            "title": "Updated Integration Post Title",
            "body": "The body content has been updated.",
            "userId": 1 # userId should generally not change, but included for completeness
        }
        try:
            updated_resource = client.update_resource("/posts", resource_id=post_to_update_id, data=updated_post_data)
            print(f"Successfully updated post {post_to_update_id}. New title: '{updated_resource.get('title')}'")
        except APIIntegrationError as e:
            print(f"Error updating post {post_to_update_id}: {e}")
    else:
        print("Skipping update test as no post was created successfully.")


    print("\n--- 5. Deleting a resource (e.g., the newly created/updated post) ---")
    if 'created_post' in locals() and created_post and created_post.get('id'):
        post_to_delete_id = created_post['id']
        try:
            client.delete_resource("/posts", resource_id=post_to_delete_id)
            print(f"Successfully deleted post {post_to_delete_id}.")
            # Verify deletion (expecting 404 or empty response)
            # This part is commented out as JSONPlaceholder doesn't actually delete resources,
            # but for a real API, you would typically verify deletion.
            # try:
            #     deleted_check = client.get_resource

This document outlines the professional output for "Step 2 of 2: projectmanager → create_project" within the "API Integration Builder" workflow. This step focuses on defining and initiating the technical project for your API integration, laying the groundwork for subsequent code generation and development.

API Integration Project Initiation: [Your Integration Name Here]

This document serves as the foundational blueprint for developing the API integration project. It details the project's scope, proposed architecture, technology stack, and initial development setup, ensuring a structured and efficient path to implementation.

1. Project Overview & Objectives

Project Name: [Your Integration Name Here]

Purpose: To establish robust and reliable connectivity between [System A - e.g., Your Internal CRM] and [System B - e.g., Salesforce, Stripe, Google API].

Primary Objectives:

• [Objective 1]: (e.g., Synchronize customer data from Salesforce to our internal CRM in real-time.)
• [Objective 2]: (e.g., Process payments securely via Stripe for e-commerce transactions.)
• [Objective 3]: (e.g., Retrieve product information from an external vendor API to update our catalog.)
• Ensure data consistency, integrity, and timely exchange between the connected systems.
• Develop a scalable, secure, and maintainable integration solution.

2. Integration Scope & Requirements

This section defines the specific functionalities and data flows that will be implemented.

2.1. External API(s) to Integrate

• API Provider: [e.g., Salesforce, Stripe, Google Maps, HubSpot]
• Base URL: [e.g., https://api.salesforce.com/services/data/vXX.X/]
• Key Endpoints/Resources:

* [Endpoint 1]: (e.g., /sobjects/Account/ - for fetching/creating accounts)

* [Endpoint 2]: (e.g., /v1/charges - for processing payments)

* [Endpoint 3]: (e.g., /maps/api/geocode/json - for geocoding addresses)

• Authentication Method: [e.g., OAuth 2.0 (Client Credentials/Authorization Code Flow), API Key in Header, Bearer Token]
• Rate Limits & Quotas: (e.g., 5,000 requests/hour, 100,000 requests/day) - To be handled with appropriate retry mechanisms and back-off strategies.

2.2. Internal Systems/Applications Involved

• System Name: [e.g., Internal CRM, E-commerce Platform, Data Warehouse]
• Interaction Method: [e.g., Internal API, Database Access, Message Queue]
• Relevant Endpoints/Schemas: (e.g., /api/v1/customers, customers table schema)

2.3. Key Use Cases / Functionalities

• Use Case 1: [Data Synchronization - e.g., Customer Sync]

* Trigger: New customer created/updated in [System A].

* Action: Fetch customer details from [System A] API, transform data, and create/update corresponding customer record in [System B] API.

• Use Case 2: [Transactional Processing - e.g., Payment Processing]

* Trigger: User initiates checkout on [E-commerce Platform].

* Action: Send payment details to [Stripe API], handle success/failure responses, update order status in [E-commerce Platform].

• Use Case 3: [Data Enrichment - e.g., Address Geocoding]

* Trigger: New address added in [Internal System].

* Action: Send address to [Google Maps Geocoding API], retrieve coordinates, update internal record.

2.4. Data Models & Transformations

• Input Data Structures (from External API): Define expected JSON/XML schemas.
• Output Data Structures (for Internal System): Define required JSON/XML schemas or database table mappings.
• Required Transformations: Specify rules for mapping fields, data type conversions, aggregations, or enrichments.

3. Proposed Technical Architecture

This section outlines the high-level design choices for the integration solution.

• Integration Pattern:

* [Select One]:

* Direct API Call: For simple, synchronous request-response interactions.

* API Gateway: For centralized management, routing, security, and transformation of multiple integrations.

* Message Queue/Event-Driven: For asynchronous, decoupled communication, enhancing resilience and scalability (e.g., RabbitMQ, Kafka, AWS SQS).

* ETL (Extract, Transform, Load): For batch processing and large-scale data synchronization.

• Deployment Model:

* [Select One]:

* Serverless Functions: (e.g., AWS Lambda, Azure Functions, Google Cloud Functions) - Ideal for event-driven, cost-effective execution.

* Containerized Microservice: (e.g., Docker, Kubernetes) - For isolated, scalable, and portable services.

* Dedicated VM/Server: For specific resource requirements or legacy environments.

• Scalability & Resilience:

* Load Balancing: Distribute requests across multiple instances.

* Retry Mechanisms: Implement exponential back-off for transient API errors.

* Circuit Breakers: Prevent cascading failures when external APIs are unresponsive.

* Idempotency: Ensure operations can be repeated without unintended side effects.

• Security Considerations:

* API Key/Secret Management: Secure storage and retrieval (e.g., AWS Secrets Manager, HashiCorp Vault, environment variables).

* Data Encryption: Encrypt sensitive data in transit (TLS/SSL) and at rest.

* Least Privilege: Grant only necessary permissions to the integration service.

* Input Validation: Sanitize and validate all incoming and outgoing data.

• Monitoring & Logging:

* Centralized Logging: Aggregate logs (e.g., ELK Stack, Splunk, CloudWatch Logs) for easy debugging and auditing.

* Metrics & Alerts: Track key performance indicators (e.g., request latency, error rates, throughput) and configure alerts for anomalies.

* Distributed Tracing: (e.g., OpenTelemetry, Jaeger) for understanding request flow across services.

4. Recommended Technology Stack

This outlines the core technologies that will be used for development.

• Programming Language: [e.g., Python 3.9+, Node.js 18+, Java 17+, Go 1.20+]
• Frameworks/Libraries:

* Python: Flask, FastAPI, Requests, Pydantic

* Node.js: Express.js, Axios, dotenv, Joi

* Java: Spring Boot, OkHttp, Jackson

* Go: net/http, gorilla/mux, gorm

• Authentication Libraries: (Specific to chosen language/framework for OAuth2, JWT handling, etc.)
• Dependency Management: [e.g., pip (Python), npm/yarn (Node.js), Maven/Gradle (Java), go mod (Go)]
• Runtime Environment: As specified by the chosen language version.
• Version Control: Git (managed via [e.g., GitHub, GitLab, Bitbucket]).

5. Initial Development Project Setup & Structure

This section outlines the foundational steps and proposed structure for the integration project, preparing for the actual code generation phase.

5.1. Version Control Repository Initialization

• Repository Creation: A new Git repository will be created on [e.g., GitHub, GitLab] named [your-integration-name]-integration.
• Initial Commit: The repository will be initialized with:

* .gitignore: To exclude development artifacts and sensitive files.

* README.md: Containing project description, setup instructions, and contribution guidelines.

* LICENSE: Specifying the project's licensing.

5.2. Local Development Environment Setup

• Prerequisites: Developers will need [Node.js/npm, Python/pip, Docker, Java JDK] installed, along with a preferred IDE (e.g., VS Code, IntelliJ IDEA).
• Virtual Environment (if applicable): A dedicated virtual environment will be created for Python/Node.js projects to manage dependencies in isolation.
• Configuration Management:

* Environment-specific variables (API keys, URLs, database credentials) will be managed using .env files for local development.

* For production, these will transition to secure environment variables or a secrets management service.

5.3. Proposed Project Directory Structure

A standardized and logical directory structure will be implemented to enhance maintainability and readability.

/project-