SDK Utilities

An overview of the different ways Lumos helps make implementation steps easier

There are many different “quality of life” utilities in the SDK. Generally most utilities can be found in the connector.utils namespace. These include utilities like:

String manipulation, for example split_name and full_name
Response utilities like create_client_response
BearerAuth for simple httpx authentication setup
JWT signing utility (sign_jwt)
A full suite of utilities for authorizing with an OAuth1 API
Pagination and its related classes
The sync_to_async decorator for handling sync method calls
Other utilities

Be sure to explore what the SDK has to offer, as it has been developed and tested with many currently deployed Lumos integrations, to have productive and effective custom connector development.

HTTP Servers

Each connector has the ability to run as a traditional HTTP API server. This is beneficial especially for local testing, since each implemented capability is surfaced as an API endpoint. To run a HTTP server from a connector, run the following:

abc hacking http-server

You can add a --reload flag, the server will reload during any changes to the connector. This also surfaces the documentation and self-describing nature of Lumos SDK connectors. You can access the docs by navigating to [http://127.0.0.1:8000](http://127.0.0.1:8000/) /docs or [http://127.0.0.1:8000](http://127.0.0.1:8000/)/redoc. Allowing you to see each capability clearly documented, and download the OpenAPI Specification. As mention above, each capability is surfaced as an API endpoint. An example of calling such a capability:

curl --location 'http://127.0.0.1:8000/list_accounts' \
--header 'Content-Type: application/json' \
--data '{
    "auth": {
        "token": {
            "token": "some_token"
        }
    },
    "settings": {},
    "request": {}
}'

Note that all communication is done via JSON and POST. The connector does not return other HTTP statuses than 200, as it describes each error or issue in it’s JSON response.

Rate Limiting

One of the common pitfalls of API integrations is rate limiting. Traditionally, each service implements a rate limiting policy with a certain amount of requests available during a time window. These instances can be problematic when performing a large amount of fetching. For such purposes the SDK provides a RateLimitedClient class. This class is tightly integrated with the above mentioned BaseIntegrationClient and extends its functionality on a per-code-block basis.

# Example: abc/client.py

ABC_RATE_LIMIT_CONFIG = RateLimitConfig(
    app_id="abc",
    requests_per_window=30,
    window_seconds=60,
    max_delay=60 * 1.2,
)

Typically you would implement a rate limiting config in your connector, that basically copies the way the end systems policy is setup. Required parameters are app_id, requests_per_window, and window_seconds. The max_delay argument in example serves as an increase in delay between calls, making the client wait longer before retrying the rate limited requests. You can now either apply this configuration to your whole AbcClient or to selective capability calls.

# Example: abc/client.py

class AbcClient(BaseIntegrationClient):
	def __init__(
	        self, args: Request, rate_limit_config: RateLimitConfig = ABC_RATE_LIMIT_CONFIG
	    ) -> None:
			    # All calls made from this client will be handled for rate limiting.
	        super().__init__(args, rate_limit_config)
	        
# Example: abc/capabilities_read.py

async def list_accounts(args: ListAccountsRequest) -> ListAccountsResponse:
    async with AbcClient(args, ABC_RATE_LIMIT_CONFIG) as client:
        # All calls within this context are being handled for rate limiting

This serves as a simple way to implement exponential backoffs - delays between requests and retries. Currently the supported strategy is FIXED which is a configuration combining sane defaults and supplied arguments.

Testing

Proper test coverage helps with connector maintenance. The scaffolded connector comes with some of the tests already prepared for finishing. To run tests for a connector, use the following command:

pytest .

There is a specific structure to tests and some amount of abstraction to make writing tests easier. Mainly the SDK uses pytest-cases to run it’s tests. Each connector has a predefined structure:

abc/
    abc/
    tests/
        test_read_capabilities/
            test_list_accounts_cases.py
            ...
        test_write_capabilities/
		        test_create_account_cases.py
		        ...
				common_mock_data.py
				test_all_capabilities.py

Each capability has its own test case file in the corresponding directory. The common_mock_data.py file is used to contain any mocks, dummy data or any other constants that you might need to be using during the testing. And lastly the test_all_capabilities.py is responsible for running all of the test case files and testing your implementation. This way the cases can be extended, and the test runner can be modified to fit specific needs of the end system.

Quickstart

Lumos Connectors

Lumos

Guides

Webhooks

Examples

Tools

Recipes

SDK Utilities

An overview of the different ways Lumos helps make implementation steps easier

HTTP Servers

Rate Limiting

Testing

​An overview of the different ways Lumos helps make implementation steps easier

​HTTP Servers

​Rate Limiting

​Testing

An overview of the different ways Lumos helps make implementation steps easier

HTTP Servers

Rate Limiting

Testing