python

api_client.py

import os

import requests

import logging

from dotenv import load_dotenv

from requests.exceptions import Timeout, ConnectionError as RequestsConnectionError, RequestException

Load environment variables from .env file

load_dotenv()

Configure logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

logger = logging.getLogger(__name__)

Import custom exceptions

from api_exceptions import (

APIError, AuthenticationError, ForbiddenError, NotFoundError,

BadRequestError, ValidationError, RateLimitError, ServerError,

ConnectionError, TimeoutError, ConflictError

)

class BaseAPIClient:

"""

A generic base client for interacting with RESTful APIs.

Handles common functionalities like HTTP requests, authentication,

error handling, and logging.

"""

def __init__(self, base_url, api_key=None, auth_type='header',

api_key_name='X-API-Key', timeout_seconds=30,

max_retries=3, backoff_factor=0.5):

"""

Initializes the BaseAPIClient.

Args:

base_url (str): The base URL of the API (e.g., "https://api.example.com/v1").

api_key (str, optional): The API key for authentication.

auth_type (str): Type of authentication.

'header': API key sent in a custom header (default).

'bearer_token': API key sent as a Bearer token in Authorization header.

'query_param': API key sent as a query parameter.

api_key_name (str): The name of the header or query parameter for the API key.

Defaults to 'X-API-Key'.

timeout_seconds (int): Default timeout for requests in seconds.

max_retries (int): Maximum number of retries for transient errors.

backoff_factor (float): Factor for exponential backoff between retries.

"""

if not base_url:

raise ValueError("Base URL cannot be empty.")

self.base_url = base_url.rstrip('/')

self.api_key = api_key

self.auth_type = auth_type

self.api_key_name = api_key_name

self.timeout_seconds = timeout_seconds

self.max_retries = max_retries

self.backoff_factor = backoff_factor

self.session = requests.Session() # Use a session for connection pooling

logger.info(f"Initialized BaseAPIClient for {self.base_url} with auth_type: {self.auth_type}")

def _get_headers(self, custom_headers=None):

"""

Constructs the request headers, including authentication.

Args:

custom_headers (dict, optional): Additional headers to include.

Returns:

dict: The combined dictionary of headers.

"""

headers = {

'Content-Type': 'application/json',

'Accept': 'application/json'

}

if custom_headers:

headers.update(custom_headers)

if self.api_key:

if self.auth_type == 'header':

headers[self.api_key_name] = self.api_key

elif self.auth_type == 'bearer_token':

headers['Authorization'] = f'Bearer {self.api_key}'

# For 'query_param', the key is added directly to the URL parameters, not headers.

return headers

def _handle_response(self, response):

"""

Handles the API response, raising custom exceptions for error status codes.

Args:

response (requests.Response): The response object from an HTTP request.

Raises:

AuthenticationError: For 401 Unauthorized.

ForbiddenError: For 403 Forbidden.

NotFoundError: For 404 Not Found.

BadRequestError: For 400 Bad Request.

ValidationError: For 422 Unprocessable Entity.

ConflictError: For 409 Conflict.

RateLimitError: For 429 Too Many Requests.

ServerError: For 5xx status codes.

APIError: For other unhandled HTTP errors.

"""

status_code = response.status_code

try:

# Attempt to parse JSON response for error details

error_details = response.json()

except requests.exceptions.JSONDecodeError:

error_details = response.text

error_message = f"API request failed with status {status_code}"

if isinstance(error_details, dict) and ('message' in error_details or 'error' in error_details):

error_message = error_details.get('message', error_details.get('error', error_message))

elif isinstance(error_details, str) and error_details:

error_message = error_details

if 200 <= status_code < 300:

return # Success, no exception

elif status_code == 400:

raise BadRequestError(error_message, status_code, error_details)

elif status_code == 401:

raise AuthenticationError(error_message, status_code, error_details)

elif status_code == 403:

raise ForbiddenError(error_message, status_code, error_details)

elif status_code == 404:

raise NotFoundError(error_message, status_code, error_details)

elif status_code == 409:

raise ConflictError(error_message, status_code, error_details)

elif status_code == 422:

raise ValidationError(error_message, status_code, error_details)

elif status_code == 429:

raise RateLimitError(error_message, status_code, error_details)

elif 500 <= status_code < 600:

raise ServerError(error_message, status_code, error_details)

else:

raise APIError(error_message, status_code, error_details)

def _request(self, method, endpoint, params=None, data=None, json=None,

headers=None, timeout=None, allow_redirects=True):

"""

Executes an HTTP request with retry logic and error handling.

Args:

method (str): The HTTP method (e.g., 'GET', 'POST', 'PUT', 'DELETE').

endpoint (str): The API endpoint (e.g., '/products').

params (dict, optional): Dictionary of query parameters.

data (dict or str, optional): Dictionary or string of form data.

json (dict, optional): Dictionary to be sent as JSON body.

headers (dict, optional): Custom headers for this specific request.

timeout (int, optional): Timeout for this specific request. Defaults to client's timeout.

allow_redirects (bool): Whether to follow redirects.

Returns:

requests.Response: The successful response object.

Raises:

APIError: For any API-related error.

ConnectionError: For network connection issues.

TimeoutError: For request timeouts.

"""

url = f"{self.base_url}{endpoint}"

full_headers = self._get_headers(headers)

request_timeout = timeout if timeout is not None else self.timeout_seconds

# Add API key to query params if auth_type is 'query_param'

if self.api_key and self.auth_type == 'query_param':

if params is None:

params = {}

params[self.api_key_name] = self.api_key

for attempt in range(self.max_retries):

try:

logger.debug(f"Attempt {attempt + 1}/{self.max_retries}: {method} {url}")

response = self.session.request(

method,

url,

params=params,

data=data,

json=json,

headers=full_headers,

timeout=request_timeout,

allow_redirects=allow_redirects

)

self

API Integration Builder: Project Creation Module

This document outlines the comprehensive output for Step 2 of 2 in the "API Integration Builder" workflow, focusing on generating robust, professional code to integrate with an external Project Management System (PMS) API for the purpose of creating new projects.

1. Introduction: Objective and Scope

The objective of this step is to deliver production-ready code that facilitates the creation of new projects within a specified external Project Management System via its API. Our "API Integration Builder" has processed your requirements and is now providing the detailed implementation for the projectmanager → create_project action.

This deliverable includes:

• A clear understanding of the integration's core components.
• The essential prerequisites for configuring and utilizing the generated code.
• An illustrative example of the generated code structure, focusing on best practices.
• Actionable steps for deploying, testing, and extending the integration.

2. Key Integration Components for Project Creation

To successfully integrate and create projects via an external API, several critical components must be addressed:

• API Endpoint & Method: The specific URL and HTTP method (e.g., POST) designated by the target PMS API for creating new projects.
• Authentication: Securely verifying the identity of your application to the PMS API. Common methods include API Keys, OAuth 2.0 tokens, or Basic Authentication.
• Request Payload Structure: The data format (typically JSON) and required fields that the PMS API expects when creating a project (e.g., project name, description, start date, team members). This structure must strictly adhere to the target API's documentation.
• Response Handling: Interpreting the API's response, including success confirmations (e.g., HTTP 201 Created, returning the new project's ID) and extracting relevant information.
• Error Handling: Gracefully managing potential issues such as invalid input, authentication failures, network errors, or API rate limits, providing informative feedback.
• Configuration Management: Externalizing sensitive information (API keys, base URLs) and frequently changing parameters for easy management and security.

3. Prerequisites for Code Generation

To ensure the generated code is tailored to your specific needs, the API Integration Builder relies on the following information, which we assume was gathered in previous steps:

• Target Project Management System (PMS) API:

* API Base URL: The root URL for all API requests (e.g., https://api.example-pms.com/v1).

* Project Creation Endpoint: The specific path for creating projects (e.g., /projects).

* HTTP Method: Usually POST for creation.

• Authentication Details:

* Authentication Type: (e.g., API Key, OAuth 2.0 Bearer Token).

* Credential Storage: How credentials will be provided (e.g., environment variables, configuration file).

• Required Project Fields: A list of mandatory and optional fields for creating a project, as per the target PMS API documentation (e.g., name, description, startDate, ownerId, teamIds).
• Preferred Programming Language/Framework: The language and potentially framework (e.g., Python, Node.js, Java, Go) in which the integration code should be generated. (For this deliverable, we will use Python as an illustrative example).

4. Generated Code Structure (Illustrative Example: Python)

Below is a conceptual example of the generated Python code. This structure is designed for modularity, readability, and ease of maintenance.

import os
import requests
import json

class ProjectManagerAPI:
    """
    Client for integrating with an external Project Management System (PMS) API.
    Handles project creation and related operations.
    """

    def __init__(self, base_url: str = None, api_key: str = None):
        """
        Initializes the ProjectManagerAPI client.

        Args:
            base_url (str): The base URL of the PMS API. Defaults to an environment variable.
            api_key (str): The API key for authentication. Defaults to an environment variable.
        """
        self.base_url = base_url or os.getenv("PMS_API_BASE_URL", "https://api.your-pms.com/v1")
        self.api_key = api_key or os.getenv("PMS_API_KEY")

        if not self.base_url:
            raise ValueError("PMS API Base URL is not provided and not found in environment variables.")
        if not self.api_key:
            raise ValueError("PMS API Key is not provided and not found in environment variables.")

        # Common headers for all requests
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"  # Example for Bearer Token
            # Add other headers if required by your PMS API (e.g., 'X-API-KEY')
        }
        self.projects_endpoint = f"{self.base_url}/projects" # Specific endpoint for projects

    def create_project(self, project_data: dict) -> dict:
        """
        Creates a new project in the external PMS.

        Args:
            project_data (dict): A dictionary containing project details.
                                 Example: {
                                     "name": "New Marketing Campaign",
                                     "description": "Plan and execute Q3 marketing initiatives.",
                                     "startDate": "2023-07-01",
                                     "endDate": "2023-09-30",
                                     "ownerId": "user-123",
                                     "status": "active"
                                 }

        Returns:
            dict: The response from the API, typically including the created project's ID and details.

        Raises:
            requests.exceptions.RequestException: For network-related errors.
            ValueError: For invalid input data or API-specific errors.
        """
        if not isinstance(project_data, dict):
            raise ValueError("project_data must be a dictionary.")
        if "name" not in project_data or not project_data["name"]:
            raise ValueError("Project 'name' is a required field.")
        # Add more validation based on your PMS API's requirements

        try:
            print(f"Attempting to create project at: {self.projects_endpoint}")
            print(f"Payload: {json.dumps(project_data)}")

            response = requests.post(
                self.projects_endpoint,
                headers=self.headers,
                json=project_data,
                timeout=10 # Set a timeout for the request
            )
            response.raise_for_status()  # Raise an HTTPError for bad responses (4xx or 5xx)

            # Assuming the API returns JSON on success
            return response.json()

        except requests.exceptions.HTTPError as e:
            error_details = {"status_code": e.response.status_code, "message": e.response.text}
            print(f"HTTP Error creating project: {error_details}")
            raise ValueError(f"API Error ({e.response.status_code}): {e.response.text}") from e
        except requests.exceptions.ConnectionError as e:
            print(f"Connection Error creating project: {e}")
            raise requests.exceptions.RequestException(f"Network connection error: {e}") from e
        except requests.exceptions.Timeout as e:
            print(f"Timeout Error creating project: {e}")
            raise requests.exceptions.RequestException(f"Request timed out: {e}") from e
        except requests.exceptions.RequestException as e:
            print(f"An unexpected request error occurred: {e}")
            raise requests.exceptions.RequestException(f"An unexpected error occurred: {e}") from e
        except json.JSONDecodeError:
            print(f"Failed to decode JSON from response: {response.text}")
            raise ValueError("Failed to parse API response as JSON.")
        except Exception as e:
            print(f"An unknown error occurred: {e}")
            raise

# --- Example Usage ---
if __name__ == "__main__":
    # IMPORTANT: Set these environment variables before running, or pass them directly
    # export PMS_API_BASE_URL="https://api.your-pms.com/v1"
    # export PMS_API_KEY="your_super_secret_api_key"

    try:
        # Initialize the API client (it will pick up credentials from env vars by default)
        pms_client = ProjectManagerAPI()

        # Define the data for the new project
        new_project_data = {
            "name": "Q4 Product Launch Planning",
            "description": "Coordinate all activities for the upcoming product launch.",
            "startDate": "2023-10-01",
            "endDate": "2023-12-15",
            "ownerId": "john.doe",
            "teamIds": ["dev-team", "marketing-team"],
            "priority": "high",
            "status": "planning"
        }

        print("\n--- Attempting to create a new project ---")
        created_project = pms_client.create_project(new_project_data)
        print("\nProject created successfully!")
        print(f"Created Project ID: {created_project.get('id', 'N/A')}")
        print(f"Full Response: {json.dumps(created_project, indent=2)}")

        # Example of handling an error (e.g., missing required field)
        print("\n--- Attempting to create a project with missing name (expected error) ---")
        invalid_project_data = {
            "description": "This project should fail.",
            "startDate": "2023-11-01"
        }
        try:
            pms_client.create_project(invalid_project_data)
        except ValueError as e:
            print(f"Caught expected error: {e}")

    except ValueError as e:
        print(f"Configuration Error: {e}")
    except requests.exceptions.RequestException as e:
        print(f"API Integration Error: {e}")
    except Exception as e:
        print(f"An unexpected error occurred during execution: {e}")

5. Actionable Steps for the Customer

Upon receiving this generated code, please follow these steps to integrate and utilize it effectively:

1. Review the Code: Carefully examine the generated code. Ensure it aligns with your understanding of the target PMS API's requirements and your project's specific needs. Pay close attention to:

* Endpoint Configuration: Verify self.projects_endpoint is correct.

* Authentication Headers: Confirm the Authorization header format matches your API's specification (e.g., Bearer, Basic, X-API-KEY).

* Request Payload: Check that the project_data structure within create_project method (and its validation) aligns with the PMS API's expected JSON body.

* Error Handling: Understand how different API responses and network issues are handled.

1. Configure Credentials:

* Environment Variables (Recommended): Set the PMS_API_BASE_URL and PMS_API_KEY (or equivalent) environment variables in your deployment environment. This is the most secure way to manage sensitive credentials.

* Direct Initialization: Alternatively, you can pass these values directly when initializing the ProjectManagerAPI class, though this is less recommended for production environments.

Example for environment variables:

    export PMS_API_BASE_URL="https://api.your-pms.com/v1"
    export PMS_API_KEY="your_actual_api_key_or_bearer_token"

1. Install Dependencies: If using Python, ensure requests library is installed:

    pip install requests

For other languages, install their respective HTTP client libraries.

1. Testing:

* Unit Tests: We recommend writing unit tests for the create_project method using mock objects for the requests library to simulate API responses (success, various errors) without making actual API calls.

Integration Tests: Perform actual calls to a development or sandbox instance* of your PMS API using the provided if __name__ == "__main__": block or by integrating the ProjectManagerAPI class into a simple script. Verify that projects are created correctly and that error handling works as expected.

1. Deployment:

* Integrate this module into your application or service.

* Ensure your deployment environment has the necessary environment variables configured for security.

* Monitor logs for any API-related errors.

1. Customization & Extension:

* Additional Fields: If your PMS API supports more fields for project creation (e.g., custom fields), modify the project_data dictionary and potentially add more validation within the create_project method.

* Other Operations: To add other API operations (e.g., get_project, update_project, delete_project), you can extend the ProjectManagerAPI class with new methods following a similar pattern.

* Logging: Enhance logging mechanisms to provide more detailed insights into API requests and responses for debugging and auditing.

6. Support and Further Assistance

Should you encounter any issues, require further customization, or wish to extend this integration, please do not hesitate to reach out to our support team. We are available to assist with:

• Troubleshooting integration issues.
• Adapting the code for specific edge cases or complex API requirements.
• Generating code for additional API endpoints or operations within the same or different external systems.
• Guidance on deployment best practices and monitoring.

Your success with this integration is our priority.