Implementation:Mage ai Mage ai Knowi Client
| Knowledge Sources | |
|---|---|
| Domains | Data_Integration, Knowi, API |
| Last Updated | 2026-02-09 00:00 GMT |
Overview
Concrete HTTP client for making authenticated requests to the Knowi API, provided by the Mage Knowi source connector.
Description
The KnowiClient class manages all HTTP communication with the Knowi API at https://knowi.com/api/1.0. It authenticates using Bearer token via access_token. The client is implemented as a context manager, verifying the access token on entry by hitting the /dashboards endpoint and closing the session on exit. Rate limiting is enforced via the Singer @utils.ratelimit(1000, 60) decorator. It uses Singer metrics.http_request_timer for request timing instrumentation. A comprehensive exception hierarchy maps HTTP status codes (400-500) to specific error classes. Configurable request timeout defaults to 300 seconds. Provides get, post, and perform convenience methods.
Usage
Used internally by Knowi stream classes. Requires access_token and optionally request_timeout and user_agent.
Code Reference
Source Location
- Repository: mage-ai
- File:
mage_integrations/mage_integrations/sources/knowi/client.py - Lines: 1-250
Signature
class KnowiClient(object):
def __init__(
self,
logger,
access_token,
config_request_timeout,
user_agent=None,
):
Import
from mage_integrations.sources.knowi.client import KnowiClient
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| logger | Logger | Yes | Logger instance for diagnostic output |
| access_token | str | Yes | Bearer token for Knowi API authentication |
| config_request_timeout | float/str | Yes | Request timeout in seconds (defaults to 300 if falsy) |
| user_agent | str | No | Optional User-Agent header value |
Outputs
| Name | Type | Description |
|---|---|---|
| response | dict | Parsed JSON response from the Knowi API |
Key Methods
| Method | Description |
|---|---|
check_access_token() |
Validates the access token by querying /dashboards. Checks for list key in response.
|
request(method, path, url, **kwargs) |
Core request method with rate limiting, metric timing, and error handling. |
get(path, **kwargs) |
Convenience wrapper for GET requests with logging. |
post(path, **kwargs) |
Convenience wrapper for POST requests. |
perform(method, path, **kwargs) |
Dispatches to get or post based on method string.
|
Error Handling
Defines a hierarchy of custom exceptions mapped to HTTP status codes (400, 401, 402, 403, 404, 405, 406, 408, 409, 415, 422, 423, 500). Falls back to KnowiError for unmapped codes. Raises Server5xxError for status codes >= 500.
Usage Examples
from mage_integrations.sources.knowi.client import KnowiClient
with KnowiClient(
logger=logger,
access_token='your_knowi_access_token',
config_request_timeout=300,
) as client:
dashboards = client.get('dashboards')
widgets = client.get('widgets')