python

api_client.py

import requests

import time

import logging

from functools import wraps

from typing import Dict, Any, Optional, Tuple, Callable

Import configuration settings

import config

Set up logging

logging.basicConfig(level=getattr(logging, config.LOG_LEVEL.upper(), logging.INFO),

format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

logger = logging.getLogger(__name__)

class ApiException(Exception):

"""Custom exception for API errors."""

def __init__(self, message: str, status_code: Optional[int] = None, response_data: Optional[Any] = None):

super().__init__(message)

self.status_code = status_code

self.response_data = response_data

def __str__(self):

return (f"API Error: {self.message} (Status: {self.status_code}, "

f"Response: {self.response_data})" if self.status_code else self.message)

def retry_on_exception(max_retries: int, initial_delay: float):

"""

Decorator for retrying a function call upon specific exceptions or HTTP status codes.

Uses exponential backoff.

"""

def decorator(func: Callable):

@wraps(func)

def wrapper(self, args, *kwargs):

retries = 0

delay = initial_delay

while retries <= max_retries:

try:

return func(self, args, *kwargs)

except (requests.exceptions.ConnectionError,

requests.exceptions.Timeout,

requests.exceptions.RequestException) as e:

logger.warning(f"Network or request error during {func.__name__}: {e}. Retrying in {delay:.2f}s...")

retries += 1

if retries > max_retries:

logger.error(f"Max retries ({max_retries}) exceeded for {func.__name__}.")

raise ApiException(f"Failed after {max_retries} retries due to network/request error.") from e

time.sleep(delay)

delay *= 2 # Exponential backoff

except ApiException as e:

# Retry specifically on server errors (5xx) or rate limits (429)

if e.status_code and (500 <= e.status_code < 600 or e.status_code == 429):

logger.warning(f"API error {e.status_code} during {func.__name__}: {e.message}. Retrying in {delay:.2f}s...")

retries += 1

if retries > max_retries:

logger.error(f"Max retries ({max_retries}) exceeded for {func.__name__}.")

raise e

time.sleep(delay)

delay *= 2

else:

raise e # Re-raise other API errors immediately

return wrapper

return decorator

class ApiClient:

"""

A robust client for interacting with an external RESTful API.

Supports API Key and OAuth 2.0 Client Credentials authentication.

"""

def __init__(self,

base_url: str = config.API_BASE_URL,

api_key: Optional[str] = config.API_KEY,

api_key_header_name: Optional[str] = config.API_KEY_HEADER_NAME,

oauth_token_url: Optional[str] = config.OAUTH_TOKEN_URL,

oauth_client_id: Optional[str] = config.OAUTH_CLIENT_ID,

oauth_client_secret: Optional[str] = config.OAUTH_CLIENT_SECRET,

oauth_scopes: Optional[list] = config.OAUTH_SCOPES,

timeout: float = config.REQUEST_TIMEOUT_SECONDS):

"""

Initializes the API client.

:param base_url: The base URL of the API.

:param api_key: The API key for authentication (if used).

:param api_key_header_name: The header name for the API key (e.g., 'X-API-KEY').

:param oauth_token_url: The URL for obtaining OAuth tokens (if used).

:param oauth_client_id: The client ID for OAuth (if used).

:param oauth_client_secret: The client secret for OAuth (if used).

:param oauth_scopes: A list of scopes for OAuth (if used).

:param timeout: Default request timeout in seconds.

"""

self.base_url = base_url

self.timeout = timeout

self._session = requests.Session()

# Authentication specific attributes

self.api_key = api_key

self.api_key_header_name = api_key_header_name

self.oauth_token_url = oauth_token_url

self.oauth_client_id = oauth_client_id

self.oauth_client_secret = oauth_client_secret

self.oauth_scopes = oauth_scopes

self._access_token = None

self._token_expires_at = 0 # Unix timestamp

# Determine authentication method based on provided credentials

self.auth_method = self._determine_auth_method()

logger.info(f"API Client initialized. Base URL: {self.base_url}, Auth method: {self.auth_method}")

def _determine_auth_method(self) -> str:

"""Determines the authentication method based on provided config."""

if self.oauth_client_id and self.oauth_client_secret and self.oauth_token_url:

return "oauth2_client_credentials"

elif self.api_key and self.api_key_header_name:

return "api_key"

else:

logger.warning("No valid authentication method configured. Proceeding without authentication.")

return "none"

def _get_api_key_headers(self) -> Dict[str, str]:

"""Returns headers for API Key authentication."""

if self.api_key and self.api_key_header_name:

return {self.api_key_header_name: self.api_key}

return {}

def _get_oauth_token(self) -> str:

"""Obtains an OAuth 2.0 access token using client credentials flow."""

if not (self.oauth_token_url and self.oauth_client_id and self.oauth_client_secret):

raise ApiException("OAuth 2.0 Client Credentials not fully configured.")

# Check if token is still valid

if self._access_token and self._token_expires_at > time.time() + 60: # Refresh 60 seconds before actual expiry

return self._access_token

logger.info("Obtaining new OAuth 2.0 access token...")

token_data = {

"grant_type": "client_credentials",

"client_id": self.oauth_client_id,

"client_secret": self.oauth_client_secret,

"scope": " ".join(self.oauth_scopes) if self.oauth_scopes else ""

}

try:

response = requests.post(self.oauth_token_url, data=token_data, timeout=self.timeout)

response.raise_for_status()

token_info = response.json()

self._access_token = token_info.get("access_token")

expires_in = token_info.get("expires_in", 3600) # Default to 1 hour if not provided

self._token_expires_at = time.time() + expires_in

if not self._access_token:

raise ApiException("Access token not found in OAuth response.")

logger.info("Successfully obtained new OAuth 2.0 access token.")

return self._access_token

except requests.exceptions.RequestException as e:

logger.error(f"Failed to obtain OAuth token: {e}")

raise ApiException(f"Could not obtain OAuth token: {e}") from e

except Exception as e:

logger.error(f"Unexpected error during OAuth token acquisition: {e}")

raise ApiException(f"Unexpected error during OAuth token acquisition: {e}") from e

def _get_headers(self) -> Dict[str, str]:

"""Constructs the full set of headers for an API request."""

headers = {

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

"Accept": "application/json"

}

if self.auth_method == "api_key":

headers.update(self._get_api_key_headers())

elif self.auth_method == "oauth2_client_credentials":

access_token = self._get_oauth_token()

headers["Authorization"] = f"Bearer {access_token}"

return headers

def _request(self, method: str, path: str, **kwargs: Any) -> Any:

"""

Internal method to make an HTTP request to

Project Deliverable: API Integration Builder - projectmanager → create_project

This document details the implementation guide for integrating with an external Project Management API to execute the create_project function. This deliverable is Step 2 of 2 in your "API Integration Builder" workflow, providing the actionable code and instructions necessary to establish this specific API connection.

1. Introduction to the API Integration Builder

The API Integration Builder is designed to streamline the process of connecting your systems with external APIs. For this specific workflow, we are focusing on generating the necessary code and guidance to interact with a Project Management System's API, enabling programmatic project creation. This integration will allow your applications to automatically provision new projects within your chosen project management platform, enhancing automation and efficiency.

2. Objective of this Integration Step: create_project

The primary objective of this step is to provide a robust and ready-to-use solution for calling the create_project endpoint of a Project Management API. This will enable your system to:

• Automate Project Creation: Automatically create new projects based on triggers from other systems (e.g., a new client onboarding, a sales deal closure).
• Standardize Project Setup: Ensure projects are created with predefined templates, default settings, and necessary initial parameters.
• Reduce Manual Effort: Eliminate the need for manual project setup within the project management tool.

3. Key Concepts for create_project Integration

To successfully create a project via API, the following core components are essential:

• API Endpoint: The specific URL where the create_project request will be sent. This is typically a POST request.
• Authentication: Mechanisms to verify your application's identity and permissions (e.g., API Key, OAuth 2.0, Bearer Token).
• Request Body Parameters: The data required to define the new project (e.g., project name, description, owner, start/end dates, status).
• Response Structure: The format and content of the API's reply, indicating success or failure, and often providing the ID or details of the newly created project.

4. Detailed Integration Guide

This guide provides a step-by-step approach to integrate the create_project functionality. We will use a generic RESTful API example, which can be adapted to specific Project Management platforms like Jira, Asana, Trello, or custom solutions.

4.1. Prerequisites

Before proceeding, ensure you have:

• API Documentation: Access to the official API documentation for your target Project Management system (crucial for exact endpoints, parameters, and authentication methods).
• API Credentials: An API Key, Access Token, or OAuth client credentials issued for your application.
• Development Environment: A suitable environment (e.g., Python, Node.js, Java) to write and execute the integration code.

4.2. Step 1: Obtain API Credentials

• Action: Log in to your Project Management system's developer portal or administrative settings.
• Action: Generate or retrieve the necessary API Key, Personal Access Token, or OAuth Client ID/Secret.
• Best Practice: Store these credentials securely (e.g., environment variables, secret management service) and never hardcode them directly into your application code.

4.3. Step 2: Define API Endpoint and Headers

Identify the base URL and the specific endpoint for creating a project. Define the necessary HTTP headers, including authentication and content type.

• Example (Python):

    import os
    import requests
    import json

    # --- Configuration (replace with your actual values) ---
    PM_API_BASE_URL = os.getenv("PM_API_BASE_URL", "https://api.example-pm.com/v1")
    PM_API_KEY = os.getenv("PM_API_KEY", "YOUR_SECURE_API_KEY") # Or a Bearer token

    CREATE_PROJECT_ENDPOINT = f"{PM_API_BASE_URL}/projects" # Common RESTful pattern

    HEADERS = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {PM_API_KEY}" # Example: Bearer token authentication
        # For API Key in header: "X-API-Key": PM_API_KEY
        # For Basic Auth: requests.auth.HTTPBasicAuth(username, password)
    }

4.4. Step 3: Construct the Request Body

Based on the API documentation, define the payload (JSON object) containing the data for the new project.

• Action: Identify all required parameters (e.g., name).
• Action: Identify any optional parameters you wish to include (e.g., description, ownerId, startDate).

• Example (Python):

    def create_project_payload(project_name: str, description: str = "", owner_id: str = None, start_date: str = None, end_date: str = None, status: str = "Active"):
        """
        Constructs the JSON payload for creating a new project.
        Adjust parameters based on your specific PM API documentation.
        """
        payload = {
            "name": project_name,
            "description": description,
            "status": status,
            # Add other required/optional fields as per API documentation
        }
        if owner_id:
            payload["ownerId"] = owner_id
        if start_date:
            payload["startDate"] = start_date # Format: "YYYY-MM-DD"
        if end_date:
            payload["endDate"] = end_date     # Format: "YYYY-MM-DD"

        return payload

    # --- Example Usage ---
    project_data = create_project_payload(
        project_name="New Client Onboarding - Project X",
        description="Automated project for onboarding client X's new integration.",
        owner_id="user123",
        start_date="2023-10-26",
        status="Planning"
    )

4.5. Step 4: Execute the API Call

Use an HTTP client library (e.g., requests in Python, fetch in JavaScript) to send the POST request with the constructed payload and headers.

• Example (Python):

    def execute_create_project(payload: dict):
        """
        Sends the POST request to the PM API to create a project.
        """
        try:
            print(f"Attempting to create project: {payload.get('name')}...")
            response = requests.post(CREATE_PROJECT_ENDPOINT, headers=HEADERS, json=payload)
            response.raise_for_status() # Raises HTTPError for bad responses (4xx or 5xx)
            return response.json()
        except requests.exceptions.HTTPError as http_err:
            print(f"HTTP error occurred: {http_err}")
            print(f"Response content: {response.text}")
            return None
        except requests.exceptions.ConnectionError as conn_err:
            print(f"Connection error occurred: {conn_err}")
            return None
        except requests.exceptions.Timeout as timeout_err:
            print(f"Timeout error occurred: {timeout_err}")
            return None
        except requests.exceptions.RequestException as req_err:
            print(f"An unexpected error occurred: {req_err}")
            return None

    # --- Execute the call ---
    # new_project_response = execute_create_project(project_data)

4.6. Step 5: Handle the API Response

Parse the API response to confirm project creation and extract any relevant information, such as the new project's ID.

• Example (Python):

    # Assuming new_project_response holds the successful JSON response
    # if new_project_response:
    #     print("\nProject created successfully!")
    #     project_id = new_project_response.get("id") # Common field for new resource ID
    #     project_url = new_project_response.get("url") # Link to the new project
    #     print(f"New Project ID: {project_id}")
    #     print(f"View Project at: {project_url}")
    # else:
    #     print("\nFailed to create project.")

4.7. Step 6: Implement Error Handling and Logging

Robust error handling is critical for production systems.

• Action: Catch specific HTTP status codes (e.g., 400 Bad Request, 401 Unauthorized, 403 Forbidden, 404 Not Found, 500 Internal Server Error).
• Action: Log errors with sufficient detail (timestamp, error type, request payload, response content) for debugging.
• Action: Implement retry mechanisms for transient errors (e.g., 429 Too Many Requests, 503 Service Unavailable).

5. Complete Example Code (Python)

This consolidated example provides a functional script for creating a project.

import os
import requests
import json
import logging

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

# --- Configuration (Load from Environment Variables for Security) ---
PM_API_BASE_URL = os.getenv("PM_API_BASE_URL", "https://api.example-pm.com/v1")
PM_API_KEY = os.getenv("PM_API_KEY", "YOUR_SECURE_API_KEY") # Replace with your actual API Key or Bearer Token

# Ensure API key is set
if PM_API_KEY == "YOUR_SECURE_API_KEY" or not PM_API_KEY:
    logging.error("PM_API_KEY environment variable not set. Please configure it.")
    exit(1)

CREATE_PROJECT_ENDPOINT = f"{PM_API_BASE_URL}/projects" # Common RESTful pattern

HEADERS = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {PM_API_KEY}" # Example: Bearer token authentication
    # Adjust authentication header based on your PM API documentation
    # e.g., "X-API-Key": PM_API_KEY, or use `auth` parameter in requests.post for Basic Auth
}

def create_project_payload(project_name: str, description: str = "", owner_id: str = None, start_date: str = None, end_date: str = None, status: str = "Active"):
    """
    Constructs the JSON payload for creating a new project.
    Adjust parameters based on your specific PM API documentation.
    """
    payload = {
        "name": project_name,
        "description": description,
        "status": status,
    }
    if owner_id:
        payload["ownerId"] = owner_id
    if start_date:
        payload["startDate"] = start_date # Format: "YYYY-MM-DD"
    if end_date:
        payload["endDate"] = end_date     # Format: "YYYY-MM-DD"

    return payload

def execute_create_project(payload: dict):
    """
    Sends the POST request to the PM API to create a project.
    """
    try:
        logging.info(f"Attempting to create project: {payload.get('name')}...")
        response = requests.post(CREATE_PROJECT_ENDPOINT, headers=HEADERS, json=payload)
        response.raise_for_status() # Raises HTTPError for bad responses (4xx or 5xx)

        logging.info(f"Project '{payload.get('name')}' created successfully!")
        return response.json()

    except requests.exceptions.HTTPError as http_err:
        logging.error(f"HTTP error occurred: {http_err} for project '{payload.get('name')}'")
        logging.error(f"Response status code: {response.status_code}")
        logging.error(f"Response content: {response.text}")
    except requests.exceptions.ConnectionError as conn_err:
        logging.error(f"Connection error occurred: {conn_err} while connecting to {CREATE_PROJECT_ENDPOINT}")
    except requests.exceptions.Timeout as timeout_err:
        logging.error(f"Timeout error occurred: {timeout_err} while creating project '{payload.get('name')}'")
    except requests.exceptions.RequestException as req_err:
        logging.error(f"An unexpected error occurred: {req_err}")
    return None

if __name__ == "__main__":
    # --- Example Project Data ---
    example_project = create_project_payload(
        project_name="Marketing Campaign Q4 2023",
        description="Plan and execute marketing activities for Q4 product launch.",
        owner_id="marketing_lead_id_abc", # Replace with an actual user ID from your PM system
        start_date="2023-10-01",
        end_date="2023-12-31",
        status="Planning"
    )

    # --- Execute the project creation ---
    new_project_details = execute_create_project(example_project)

    if new_project_details:
        logging.info("\n--- Project Creation Summary ---")
        logging.info(f"Project Name: {new_project_details.get('name', 'N/A')}")
        logging.info(f"Project ID: {new_project_details.get('id', 'N/A')}")
        logging.info(f"Status: {new_project_details.get('status', 'N/A')}")
        logging.info(f"Link: {new_project_details.get('url', 'Not provided')}") # Many APIs return a direct link
        logging.info("------------------------------")
    else:
        logging.error("Failed to retrieve details for the newly created project.")

6. Testing and Validation

• Unit Testing: Test the create_project_payload function with various inputs to ensure the JSON structure is correct.
• Integration Testing: Execute the execute_create_project function in a test environment (or with mock API responses) to verify the entire flow.
• Live Testing: Once confident, run the script against your actual Project Management API to confirm a project is created successfully and appears correctly in the PM system.
• Error Scenarios: Test with invalid credentials, missing required parameters, and other error conditions to ensure robust error handling.

7. Next Steps

1. Review and Adapt: Carefully review the provided code and adapt the PM_API_BASE_URL, PM_API_KEY, CREATE_PROJECT_ENDPOINT, HEADERS, and the create_project_payload function parameters to match your specific Project Management API documentation.
2. Secure Credentials: Ensure your PM_API_KEY (or other credentials) are loaded from secure environment variables or a secret management service, not hardcoded