Python SDK

Flare has a Python SDK available on PyPI. It is a light wrapper around requests that automatically manages API authentication.

Links

• Github Repository
• PyPI Page

Installing

pip install flareio

Basic Usage

The flareio package provides the FlareApiClient class, which automatically manages authentication and exposes get, post, put, and delete methods.

Basic Usage Example

import os

from flareio import FlareApiClient


client = FlareApiClient(
    api_key=os.environ["FLARE_API_KEY"],
    tenant_id=None,  # Use my default tenant.
)

resp = client.get("/tokens/test")

print(resp.json())

Creating a client from environment variables

The Flare SDK automatically recognizes the FLARE_API_KEY and FLARE_TENANT_ID environment variables.

For example, you may expose the following environment variables, and then create an API client from them:

export FLARE_API_KEY="<api-key>"
export FLARE_TENANT_ID="<tenant-id>"

from flareio import FlareApiClient


client = FlareApiClient.from_env()

resp = client.get("/tokens/test")

print(resp.json())

Paging Util - Generic

The FlareApiClient has a scroll method that can be used with endpoints that support the Flare standard paging pattern .

Paging Util - Generic Example

import os

from flareio import FlareApiClient
from flareio.ratelimit import Limiter


api_key = os.environ.get("FLARE_API_KEY")
if not api_key:
    raise Exception("Please provide an API key")

api_client = FlareApiClient(api_key=api_key)

limiter = Limiter.from_seconds(0.25)

last_from: str | None = None
fetched_pages: int = 0

for resp in api_client.scroll(
    method="GET",
    url="/astp/v2/sources",
    params={
        "from": last_from,
    },
):
    # Rate limiting (default).
    limiter.tick()

    # Get results from the response
    resp_data = resp.json()
    items = resp_data.get("items")

    fetched_pages += 1
    print(f"Fetched page {fetched_pages} ({last_from=}) with {len(items)} items...")

    # Save the last "next" value.
    last_from = resp_data.get("next") or last_from

print("The last value for 'next' was", last_from)

Paging Util - Event Feeds

The FlareApiClient has a scroll_events method that can be used with event feeds endpoints. The advantage of this method over scroll is that it also automates the fetching of individual events.

Paging Util - Event Feeds Example

import datetime

from flareio import FlareApiClient


api_client = FlareApiClient.from_env()
initial_cursor = None

from_timestamp: str = (
    datetime.datetime.now(tz=datetime.timezone.utc) - datetime.timedelta(hours=1)
).isoformat()

for result in api_client.scroll_events(
    method="POST",
    pages_url="/firework/v4/events/global/_search",
    events_url="/firework/v2/activities/",
    json={
        "size": 10,
        "order": "asc",
        "from": initial_cursor,
        "filters": {
            "type": ["chat_message"],
            "estimated_created_at": {"gte": from_timestamp},
        },
        "query": {
            "type": "query_string",
            "query_string": "hello",
        },
    },
):
    print(f"Full event data: {result.event}")
    print(f"The next execution could resume using from={result.next}")

Using a Proxy

To route requests through an HTTP proxy, pass a proxies dict to FlareApiClient. It accepts the same format as the requests proxies argument.

Using a Proxy

from flareio import FlareApiClient


client = FlareApiClient(
    api_key="fw_...",
    proxies={"https": "http://proxy.corp.local:8080"},
)

The HTTP_PROXY and HTTPS_PROXY environment variables are also picked up by requests, so setting them is enough if you’d rather not pass proxies explicitly.

Custom Session

The FlareApiClient can be initialized with a custom requests.Session. This allows, for example, to inject custom retry configuration.

Custom Session

import requests

from flareio import FlareApiClient
from requests.adapters import HTTPAdapter
from urllib3.util import Retry


session = requests.Session()
retries = Retry(
    total=5,
    backoff_factor=2,
    status_forcelist=[429, 502, 503, 504],
    allowed_methods={"GET", "POST"},
    backoff_max=15,
)
session.mount(
    "https://",
    HTTPAdapter(
        max_retries=retries,
    ),
)

api_client = FlareApiClient(
    api_key="fw_...",
    session=session,
)