bosa-connectors-binary 0.0.10

BOSA Connectors

BOSA API SDK (Bosa Connector)

A Python SDK for seamlessly connecting to APIs that implement BOSA's Plugin Architecture under HTTP Interface. This connector acts as a proxy, simplifying the integration with BOSA-compatible APIs.

Features

• Simple and intuitive API for connecting to BOSA-compatible services
• Automatic endpoint discovery and schema validation
• Built-in authentication support (BOSA API Key and User Token)
• User management and OAuth2 integration flow support
• Type-safe parameter validation
• Flexible parameter passing (dictionary or keyword arguments)
• Retry support for requests that fail (429 or 5xx)
• Response fields filtering based on action and output

Installation

Prerequisites

• Python 3.11+ - Install here
• Pip (if using Pip) - Install here
• Poetry 1.8.1+ (if using Poetry) - Install here
• Git (if using Git) - Install here
• For git installation:
  • Access to the GDP Labs SDK github repository

1. Installation from Pypi

Choose one of the following methods to install the package:

Using pip

pip install bosa-connectors-binary

Using Poetry

poetry add bosa-connectors-binary

2. Development Installation (Git)

For development purposes, you can install directly from the Git repository:

poetry add "git+ssh://[e-mail]/GDP-ADMIN/bosa-sdk.git#subdirectory=python/bosa-connectors"

Quick Start

Here's a simple example of how to use the BOSA Connector with API key authentication and user token.

Initialization

Before using the connector, you need to initialize it with your BOSA API base URL and API key.

from bosa_connectors.connector import BosaConnector

# Initialize the connector
bosa = BosaConnector(api_base_url="YOUR_BOSA_API_BASE_URL", api_key="YOUR_API_KEY")

Authentication

After initializing the connector, you can authenticate with your BOSA API key.

# User token from authentication
user_token = "Enter your key here from authentication, or refer to User Authentication section below"

# Check if a user has an integration for a connector
has_integration = bosa.user_has_integration("github", user_token)

if not has_integration:
    # Initiate the OAuth2 flow for a connector
    auth_url = bosa.initiate_connector_auth("github", user_token, "https://your-callback-uri.com")
    # Redirect the user to auth_url to complete authentication, we exit here.
    print("Integration with GitHub not found.")
    print(f"Please visit {auth_url} to complete authentication.")
    exit()

Alternatively, you can authenticate the user first and then use their token:

user = bosa.authenticate_bosa_user("username", "password")

# Get user token
user_token = user.token

Basic Example (Direct Execution)

It is the basic way to execute actions, where you need to provide the connector name, action name, and user token. The response will contain the data and status:

# Prepare input parameters
params = {
    "owner": "gdp-admin",
    "author": "samuellusandi",
    "per_page": 1,
    "sort": "author_date",
    "created_date_start": "2025-02-01",
    "created_date_end": "2025-02-02"
}

# Execute the action with user token
data, status = bosa.execute("github", "search_commits", token=user_token, max_attempts=1, input_=params)

# Print the result
print(data)
print(status)

Alternative Approach (Fluent Interface)

For more complex scenarios or more control over the execution, you can use the fluent interface. We're recommending this approach if you:

• Need to execute multiple actions with different parameters
• Expecting list response
• Need to execute actions in a loop

# Prepare input parameters
params = {
    "owner": "gdp-admin",
    "author": "samuellusandi",
    "per_page": 1,
    "sort": "author_date",
    "created_date_start": "2025-02-01",
    "created_date_end": "2025-02-02"
}

# Create a connector instance to a service
github = bosa.connect('github')

# Execute actions with fluent interface
response = github.action('list_pull_requests')\
    .params(params)\
    .max_attempts(3)\
    .token('user-token')\
    .run()  # Execute and return ActionResponse for advanced data handling

# Get initial data
initial_data = response.get_data()

# Iterate the following next pages
while response.has_next():
    response = response.next_page()
    data = response.get_data()
    # Process data here
    ...

# You can also navigate backwards
while response.has_prev():
    response = response.prev_page()
    data = response.get_data()
    # Process data here
    ...

# Execute multiple independent actions using the same connector instance
commits_response = github.action('list_commits')\
    .params({
        'owner': 'GDP-ADMIN',
        'repo': 'bosa',
        'page': 1,
        'per_page': 10
    })\
    .token('user-token')\
    .run()

run method also available for direct execution from connector instance, without using fluent interface.

# Prepare input parameters
params = {
    "owner": "gdp-admin",
    "author": "samuellusandi",
    "per_page": 1,
    "sort": "author_date",
    "created_date_start": "2025-02-01",
    "created_date_end": "2025-02-02"
}

# Execute actions with run method
response = bosa.run('github', 'list_pull_requests', params)
print(response.get_data())

Working with Files using ConnectorFile

When working with APIs that require file uploads or return file downloads, use the ConnectorFile model:

from bosa_connectors.models.file import ConnectorFile

# For uploads: Create a ConnectorFile object
with open("document.pdf", "rb") as f:
    upload_file = ConnectorFile(
        file=f.read(),
        filename="document.pdf",
        content_type="application/pdf"
    )

params = {
  "file": upload_file,
  "name": "My Document"
}

# Include in your parameters
result, status = bosa.execute("google_drive", "upload_file", input_=params)

# For downloads: Check response type
file_result, status = bosa.execute("google_drive", "download_file", input_={"file_id": "123"})
if isinstance(file_result, ConnectorFile):
    # Save to disk
    with open(file_result.filename or "downloaded_file", "wb") as f:
        f.write(file_result.file)

Available Methods

Connector Instance Methods

The connector instance provides several methods for configuring and executing actions:

• connect(name): Create a connector instance to a service

• action(name): Specify the action to execute

• params(dict): Set request parameters (including pagination parameters like page and per_page). Note that params for each plugin and action could be different

• token(str): Set the BOSA user token

• headers(dict): Set custom request headers

• max_attempts(number): Set the maximum number of retry attempts (default: 1) Execution Methods:

• run(): Execute and return ActionResponse for advanced data handling

• execute(): Execute and return data and status for basic data handling. The data part of the return value can be a ConnectorFile object when the API returns a non-JSON response (such as a file download).

Response Handling (ActionResponse)

The ActionResponse class provides methods for handling the response and pagination:

• get_data(): Get the current page data (returns the data field from the response). This can return a ConnectorFile object when the API returns a non-JSON response (such as a file download).
• get_meta(): Get the metadata information from the response (e.g., pagination details, total count)
• get_status(): Get the HTTP status code
• is_list(): Check if response is a list
• has_next(): Check if there is a next page
• has_prev(): Check if there is a previous page
• next_page(): Move to and execute next page
• prev_page(): Move to and execute previous page
• get_all_items(): Get all items from all pages (returns a list of objects containing data and meta for each page)

Data Models

The SDK uses the following data models:

• ActionResponseData: Contains the response data structure with data (list, object, or ConnectorFile instance) and meta (metadata) fields
• InitialExecutorRequest: Stores the initial request parameters used for pagination and subsequent requests
• ConnectorFile: Represents a file in requests and responses with these properties:
  • file: Raw bytes content of the file
  • filename: Optional name of the file
  • content_type: Optional MIME type of the file
  • headers: Optional HTTP headers for the file

Configuration Parameters

• api_base_url: The base URL of your BOSA API endpoint (default: "https://api.bosa.id"). This parameter is extremely important as it determines the URL of the BOSA API you are connecting to, and it will be used to populate the available actions/endpoints and their parameters upon initialization.
• api_key: Your BOSA API key for authentication. This is different from plugin-specific API keys, which are managed separately by the BOSA system.

Execution Parameters

• connector: The name of the connector to use. This parameter is used to determine the connector's available actions and their parameters.
• action: The name of the action to execute. This parameter is automatically populated by the connector based on the available actions and their parameters. The list of available actions per connector can be found in https://api.bosa.id/docs and are populated through https://api.bosa.id/connectors.
• max_attempts: The maximum number of attempts to make the API request. If the request fails, the connector will retry the request up to this number of times. The default value is 1 if not provided.
  • The retries are handled automatically by the connector, with exponential backoff.
  • The retries are only done for errors that are considered retryable (429 or 5xx).
• input_: The input parameters for the action. This parameter is a dictionary that contains the parameters for the action. The connector will validate the input parameters against the action's schema.
  • To filter response fields, simply add the response_fields parameter to the input dictionary. This parameter is a list of field names that will be returned in the response. For nested fields, you can use dot notation, e.g. user.login will return the following:

  {
    "user": {
      "login": "userlogin"
    }
  }
  

• token: Optional BOSA User Token for authenticating requests. When provided, the connector will include this token in the request headers. This is required for user-specific actions or when working with third-party integrations.

How It Works

1. Initialization: When you create a BosaConnector instance, and trigger an execute(), the connector will first populate and cache the available actions and their parameters. This is done automatically.

2. Action Discovery: The connector expects the BOSA API to expose an endpoint that lists all available actions and their parameters. This is handled automatically by BOSA's HTTP Interface.

3. Execution: When you call execute(), the connector:

   • Validates your input parameters against the action's schema
   • Handles authentication
   • Makes the API request
   • Returns the formatted response

Compatibility

While primarily tested with BOSA's HTTP Interface, this connector should theoretically work with any API that implements BOSA's Plugin Architecture, as long as it:

1. Exposes an endpoint listing available actions and their parameters
2. Follows BOSA's standard HTTP Interface specifications (through the Plugin Architecture)
   • All actions must be exposed as POST endpoints.
3. Implements the required authentication mechanisms

Error Handling

The connector includes built-in error handling for:

• Invalid parameters
• Authentication failures
• Connection issues
• API response errors

User Authentication

The BOSA Connector supports user-based authentication which allows for user-specific actions and third-party integrations:

# Create a new BOSA user
user_data = bosa.create_bosa_user("username")
# Save the secret for later use
user_secret = user_data.secret

# Authenticate a user and get their token
token = bosa.authenticate_bosa_user("username", user_secret)

# Get user information
user_info = bosa.get_user_info(token.token)

Integration Management

The BOSA Connector provides methods to manage third-party integrations for authenticated users:

# Check if a user has an integration for a connector
has_integration = bosa.user_has_integration("github", user_token)

# Initiate the OAuth2 flow for a connector
auth_url = bosa.initiate_connector_auth("github", user_token, "https://your-callback-uri.com")
# Redirect the user to auth_url to complete authentication

# Remove an integration
remove_result = bosa.remove_integration("github", user_token)

References

Product Requirements Documents(PRD):

• BOSA Connector - Product Document

Architecture Documents:

• BOSA Connector - Architecture Document

Design Documents:

• BOSA Connector - Design Document

Implementation Documents:

• BOSA Connector - Implementation Document