Skip to main content

Asynchronous client for the SPOT device API

Project description

python-spot-client

CI PyPI

An asynchronous Python client for retrieving message feeds from the SPOT device API.

Features

  • Simple asynchronous wrapper around the SPOT device feed endpoints, with optional synchronous entry points for quick scripts.
  • Data classes that model devices, feeds, and messages returned by the API.
  • Helpers for retrieving the latest messages, paginated history, or data within a time window.

Installation

Install the package from the repository root in your current environment:

pip install .

For local development install the runtime dependencies together with the documentation, test, and linting extras:

pip install -e .[docs,test,lint]

Or use the convenience dev requirements file:

pip install -r requirements-dev.txt

Quick start

To authenticate requests, pass one or more (feed_id, password) tuples when creating the client.

The client wraps the SPOT REST API and exposes models that make it easier to inspect devices, feeds, and messages. The example below fetches the latest messages from a feed and prints each device's most recent location.

import asyncio
from spot_api import SpotApi


async def main():
    feed_list = [("<your-feed-id>", "<feed-password>")]

    async with SpotApi(feed_list) as client:
        await client.request_latest()

        for message in client.feed.get_latest_messages():
            print(
                f"Device {message.messenger_name} "
                f"({message.messenger_id}) at {message.get_location()} "
                f"on {message.date_time}"
            )


if __name__ == "__main__":
    asyncio.run(main())

Usage

For scripts that do not already run an event loop, call the synchronous wrappers instead:

with SpotApi([("<your-feed-id>", "<feed-password>")]) as client:
    result = client.request_latest_sync()

Or manage lifecycle manually:

client = SpotApi([("<your-feed-id>", "<feed-password>")])
result = client.request_latest_sync()
client.close()

The synchronous helpers still rely on the underlying asynchronous client. The context manager handles teardown automatically, and the explicit close method runs the same asynchronous shutdown under the hood. If you provide your own httpx.AsyncClient, close and the context manager will leave it alone so you can manage its lifecycle yourself.

Handling SPOT throttling

SPOT requests that clients wait at least 2.5 minutes (150 seconds) between requests for the same feed and a couple of seconds when switching between different feeds. The client tracks these windows automatically and will return a SpotRequestResult with status SpotRequestStatus.THROTTLED when you should pause. Inspect the accompanying wait_seconds value to determine how long to sleep before retrying:

import asyncio

from spot_api import SpotApi, SpotRequestStatus


async def main() -> None:
    async with SpotApi([("<feed-id>", "<password>")]) as client:
        result = await client.request()
        if result.status is SpotRequestStatus.THROTTLED:
            await asyncio.sleep(result.wait_seconds)


if __name__ == "__main__":
    asyncio.run(main())

When the SPOT API responds with HTTP 429, the same status is returned together with an estimated wait. The exponential backoff used by SpotApi.request_all_pages will never exceed the documented 150 second window.

Requesting paginated history

To fetch paginated historical results instead of just the latest message, call SpotApi.request and pass the page number you need:

await client.request(page=2)

You can also iterate every available page with retry-aware pagination. The iterator yields a SpotRequestResult after each request, allowing you to react to throttling events before the next page is requested:

async for result in client.request_all_pages(merge_existing=True):
    if result.status is SpotRequestStatus.THROTTLED:
        await asyncio.sleep(result.wait_seconds)
        continue

    if result.succeeded:
        # Feed data is merged into client.feed automatically.
        print(f"Fetched page with {len(client.feed.get_latest_messages())} messages")

Requesting a date range

If you know the exact time window to inspect, provide start and end datetimes using request_with_dates:

import datetime

start = datetime.datetime(2023, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc)
end = datetime.datetime(2023, 1, 31, 23, 59, 59, tzinfo=datetime.timezone.utc)
await client.request_with_dates(start, end)

The SpotFeed object associated with the SpotApi instance is populated after any request. Inspect its devices dictionary or use helper methods such as get_latest_messages, iter_messages, or iter_messages_by_type to work with the returned data.

Interpreting request results

Every request helper returns a SpotRequestResult object containing:

  • status: the high-level result state (for example success or throttled).
  • succeeded: convenience boolean equivalent to a successful status.
  • wait_seconds: recommended wait time when requests are throttled.

This makes it straightforward to branch on status without parsing exceptions or raw HTTP responses.

Working with feeds and messages

SpotFeed and SpotMessage provide several convenience helpers for common inspection patterns:

  • SpotFeed.get_latest_messages() returns a list of the newest message per device.
  • SpotFeed.iter_messages() yields every message across the feed in reverse-chronological order, while iter_messages_by_type() filters by the message type string.
  • SpotMessage.get_location() formats the message's coordinates into a human-readable string (for example, "34.05, -118.25, 100").

These utilities make it easier to work with the deserialized data classes without needing to remember the raw JSON structure returned by SPOT.

Documentation

Project documentation is generated with Sphinx using the inline module docstrings. Install the documentation extras and build the HTML output locally:

pip install -e .[docs]
cd docs
make html

or

python -m sphinx -b html docs/ docs/build

The generated site will be available under docs/build/index.html and includes a short quick-start tour alongside the full API reference.

API references

Development

This project requires Python 3.11 or newer. After installing the test and lint extras (pip install -e .[test,lint]), run the unit tests with:

pytest

The project metadata and build configuration live in pyproject.toml. Static analysis helpers (mypy and ruff) are also included in the lint optional dependency group.

Security notes

  • Treat feed passwords like credentials; avoid committing them to source control.
  • Prefer environment variables or a local secrets manager for production usage instead of hard-coding feed credentials directly in scripts.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

python_spot_client-0.1.0.tar.gz (25.5 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

python_spot_client-0.1.0-py3-none-any.whl (19.5 kB view details)

Uploaded Python 3

File details

Details for the file python_spot_client-0.1.0.tar.gz.

File metadata

  • Download URL: python_spot_client-0.1.0.tar.gz
  • Upload date:
  • Size: 25.5 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.13.3

File hashes

Hashes for python_spot_client-0.1.0.tar.gz
Algorithm Hash digest
SHA256 84b62b25190fbf310dd999354c1a8ce0a686ea6c504d0f344595ee90f4b6866b
MD5 f5af6c9c4ecdc8cd4425cac8c6c3c322
BLAKE2b-256 bc30c676731babc6500e6cfd8aa31673dc814e851578dae82066d8003e563a57

See more details on using hashes here.

File details

Details for the file python_spot_client-0.1.0-py3-none-any.whl.

File metadata

File hashes

Hashes for python_spot_client-0.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 6190b65ce562f0998b7bfd38100a9df7cac207cc5f49480b6065d181be134508
MD5 1043b6b8cc57a8d8a2b833c2dfa88ad7
BLAKE2b-256 2555c440786177d9acbfe716fad924640033e04510a0a3abcc0b9b3087bda1d3

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page