Connections¶

The Konigle SDK provides functionality to manage third-party API integrations and connections. Connections represent authenticated links to external services using OAuth or API key authentication, storing credentials, tokens, and metadata for seamless integration with external platforms.

Overview¶

Connections are site-specific integrations that store:

Authentication credentials (API keys, OAuth tokens)
OAuth refresh tokens and expiration data
Authorization scopes and permissions
Provider-specific metadata and settings

The SDK provides read-only access to connections for security reasons. Connection creation and management must be done through the Konigle admin interface.

Listing Connections¶

Basic Listing¶

import konigle

client = konigle.Client(api_key="your-api-key")

# List all connections
connections = client.connections.list()

print(f"Found {connections.count} connections")
for connection in connections.payload:
    print(connection)

Getting Connection Credentials¶

For accessing sensitive credential data, use the get_credentials method with a provider code:

# Get credentials for a specific provider
try:
    credentials = client.connections.get_credentials("konigle_pay")

    # Credentials contain provider-specific authentication data
    print(credentials)
except Exception as e:
    print(f"Error getting credentials: {e}")

CLI Usage¶

Use the CLI for quick connection management:

# List all connections
konigle connections list

# List with pagination
konigle connections list --page 1 --page-size 5

# Get credentials for a provider
konigle connections get-credentials google

Async Operations¶

import asyncio
import konigle

async def manage_connections():
    async with konigle.AsyncClient(api_key="your-api-key") as client:
        # List connections asynchronously
        connections = await client.connections.list()

        print(f"Found {connections.count} connections")

        # Check each connection's health
        for connection in connections.payload:
            print(f"\nChecking {connection.provider}...")

            if connection.is_active:
                try:
                    # Get credentials for active connections
                    creds = await client.connections.get_credentials(connection.provider)
                    print(f"✓ {connection.provider}: Active with {len(creds)} credential fields")
                except Exception as e:
                    print(f"✗ {connection.provider}: Error accessing credentials - {e}")
            else:
                print(f"⚠️  {connection.provider}: Inactive ({connection.status})")

# Run async operation
asyncio.run(manage_connections())

Error Handling¶

from konigle.exceptions import APIError, NotFoundError

try:
    # List connections
    connections = client.connections.list()

    # Get credentials for a provider
    creds = client.connections.get_credentials("shopify")

except NotFoundError:
    print("Provider not found or no connection exists")
except APIError as e:
    print(f"API error: {e.message}")
    if e.status_code == 403:
        print("Access denied - check API key permissions")
except Exception as e:
    print(f"Unexpected error: {e}")

Security Considerations¶

Credential Access: The get_credentials method returns sensitive data including access tokens and API keys. Always handle this data securely.
Token Expiration: OAuth tokens may expire. Check is_expired and token_expires_at properties.
Read-Only Access: The SDK provides read-only access for security. Use the Konigle admin interface for connection setup and management.
Storage: Don't store the credentials. Use the api call to fetch them when needed and cache them if required.