TagLogger Python SDK

Official Python SDK for the TagLogger API — read your tags and their locations, sync your whole fleet incrementally, manage geofences, webhooks and share links, and verify incoming webhook deliveries.

It also bundles taglogger_sdk.analytics, a pure-math, client-side location-analytics toolkit (conditioning, stops/trips, dwell, clustering, GeoJSON/polyline shaping, battery projection) that runs entirely in-process over the location data the API returns — no extra services, no network calls.

• Zero runtime dependencies. Standard library only.
• Typed. Ships py.typed; full type hints on the public surface.
• Synchronous and thread-safe to reuse: a single client instance carries no mutable per-call state.
• Python 3.9+.

Install

pip install taglogger-sdk

The distribution installs as taglogger-sdk; you import it as taglogger_sdk.

Quick start

import os

from taglogger_sdk import TagLoggerClient

client = TagLoggerClient(api_key=os.environ["TAGLOGGER_API_KEY"])

# One page of tags
page = client.tags.list(limit=50)
for tag in page["data"]:
    print(tag["id"], tag.get("name"))

# Or stream every tag, paging transparently
for tag in client.tags.iterate():
    print(tag["id"])

# Latest location for one tag
location = client.tags.location("tag-1")
print(location["data"], "recording paused:", location["recordingPaused"])

Get an API key from your TagLogger dashboard under Settings → API Keys. Keys are scoped; each method below notes the scope it needs.

Configuration

client = TagLoggerClient(
    api_key=os.environ["TAGLOGGER_API_KEY"],  # never hardcode the key
    base_url="https://api.taglogger.com",  # override to target a staging deployment
    auth_header="bearer",                   # "bearer" (default) or "x-api-key"
    timeout_ms=30000,
    max_retries=2,                          # GET/DELETE retried on 5xx/network; any method retried on 429
    max_response_bytes=25 * 1024 * 1024,    # cap on buffered response bytes (default 25 MiB)
    default_headers={"X-Trace": "abc"},
    user_agent="my-app/1.0",
)

Retries. Transient failures are retried with randomized backoff: network errors and 5xx on idempotent methods (GET/DELETE), and 429 on any method (honoring Retry-After). Non-idempotent writes are never silently retried after a transport failure.

Resources

Every list endpoint returns a { "data": [...], "nextCursor": str | None } envelope and has a matching iterate* helper that auto-paginates.

Tags — client.tags

client.tags.list(limit=None, cursor=None)          # read:tags
client.tags.iterate(limit=None)                    # read:tags  -> Paginator[Tag]
client.tags.get("tag-1")                           # read:tags
client.tags.update("tag-1", {                      # manage:tags
    "alertsPause": {"paused": True, "until": None},
    "recordingPause": {"paused": True, "until": 1790000000000, "reason": "Maintenance"},
})
client.tags.battery("tag-1")                       # read:tags
client.tags.location("tag-1")                      # read:locations -> { data, recordingPaused }
client.tags.history("tag-1", start=None, end=None, order="asc", limit=None, cursor=None)  # read:locations
client.tags.iterate_history("tag-1", start=None, end=None, order="asc", limit=None)       # read:locations

Tags returned by list, iterate, get, and update include lastLocation: None unless the API key also has read:locations.

start / end accept an ISO 8601 string or epoch milliseconds. Location points expose optional endTimestampMs and sampleCount fields when a point represents a window of identical sightings; a plain point uses its own timestampMs and a count of 1.

Fleet — client.fleet

Incremental, whole-fleet location sync. Pass the high-water mark from your last sync as since (exclusive); omit it for a full sync.

client.fleet.delta(since=None, limit=None, cursor=None)   # read:locations
client.fleet.iterate_delta(since=None, limit=None)        # read:locations -> Paginator[FleetDeltaItem]

Geofences — client.geofences

client.geofences.list(limit=None, cursor=None)     # read:geofences
client.geofences.iterate(limit=None)               # read:geofences
client.geofences.get("gf-1")                       # read:geofences
client.geofences.list_integrations()                # read:geofences -> redacted binding IDs
client.geofences.create({                          # manage:geofences; recipients also need read:locations
    "name": "Yard",
    "center": {"lat": 37.77, "lng": -122.42},
    "radiusMeters": 150,
    "targetTagIds": ["tag-1", "tag-2"],            # or ["all"] for all-tags keys
    "alerts": {
        "onEntry": True,
        "emailEnabled": True,
        "emailAddresses": ["dispatch@example.com"],
        "webhookTargetIds": ["existing-slack-integration-id"],
    },
})
client.geofences.update("gf-1", {"radiusMeters": 200})    # manage:geofences
client.geofences.delete("gf-1")                    # manage:geofences -> { id, deleted }

Webhooks — client.webhooks

client.webhooks.list(limit=None, cursor=None)      # manage:webhooks
client.webhooks.get("wh-1")                        # manage:webhooks
created = client.webhooks.create({                 # manage:webhooks
    "url": "https://example.com/hooks/taglogger",
    "events": ["geofence.entry", "geofence.exit", "tag.offline"],
})
secret = created["signingSecret"]                  # shown ONCE — store it now
client.webhooks.update("wh-1", {"active": False})  # manage:webhooks
client.webhooks.rotate_secret("wh-1")              # manage:webhooks -> new signingSecret (once)
client.webhooks.send_test("wh-1")                  # manage:webhooks
client.webhooks.deliveries("wh-1", limit=None, cursor=None)   # manage:webhooks
client.webhooks.iterate_deliveries("wh-1")         # manage:webhooks
client.webhooks.delete("wh-1")                     # manage:webhooks -> { id, deleted }

The six event types are geofence.entry, geofence.exit, geofence.dwell, tag.moved, tag.battery_low, tag.offline.

Share links — client.share_links

client.share_links.list(tag_id=None, limit=None, cursor=None)  # read:share-links
client.share_links.iterate(tag_id=None, limit=None)            # read:share-links
client.share_links.get("sl-1")                                 # read:share-links
client.share_links.create({"tagId": "tag-1", "expiresInSeconds": 86400})  # manage:share-links
client.share_links.revoke("sl-1")                              # manage:share-links -> { id, revoked, status }

Customer Pages — client.customer_pages

client.customer_pages.list(limit=None, cursor=None)             # read:customer-pages
client.customer_pages.get("page-1")                            # read:customer-pages
client.customer_pages.create({                                  # manage:customer-pages
    "pageName": "Operations",
    "tagIds": ["tag-1", "tag-2"],
    "expiresAt": None,
    "features": {"liveMap": True, "zones": True, "statusTable": True, "zoneActivity": True},
})
client.customer_pages.update("page-1", {"pageName": "North fleet"})
client.customer_pages.rotate_token("page-1")                   # old /p/ URL stops resolving
client.customer_pages.revoke("page-1")
client.customer_pages.get_branding()
client.customer_pages.update_branding({"accentColor": "#0f4c81"})

Account — client.account

client.account.get()             # any valid key -> Account
client.account.usage(days=30)    # any valid key -> per-key request counts (max 40 days)

Pagination

iterate* methods return a Paginator. Iterate it for individual items, call .pages() for a page at a time, or .all() to collect everything. Pages are fetched lazily, so a large collection never has to fit in memory.

pager = client.tags.iterate(limit=100)

for tag in pager:           # individual items, across all pages
    ...

for page in pager.pages():  # one page (list) at a time
    ...

every = pager.all()         # eager list of every item

Verifying incoming webhooks

Every delivery carries an X-TagLogger-Signature header (sha256=<hex>, an HMAC-SHA256 of the raw request body keyed by the webhook's signing secret). Verify against the raw bytes before trusting the payload — re-serializing the JSON changes the bytes and fails the check.

from taglogger_sdk import (
    construct_webhook_event,
    verify_webhook_signature,
    SIGNATURE_HEADER,
    TagLoggerError,
)

# Flask example
@app.post("/hooks/taglogger")
def handle():
    raw = request.get_data()                      # raw bytes, not request.json
    signature = request.headers.get(SIGNATURE_HEADER)
    try:
        event = construct_webhook_event(raw, signature, WEBHOOK_SECRET)
    except TagLoggerError:
        return "", 400                            # invalid_signature or invalid_body
    # event is the parsed payload, e.g. {"event": "geofence.entry", ...}
    return "", 200

verify_webhook_signature(raw_body, signature, secret) -> bool is the lower-level check; it returns False (never raises) on a missing header or mismatch.

Error handling

Any non-2xx response, or a transport failure, raises TagLoggerError:

from taglogger_sdk import TagLoggerError

try:
    client.tags.get("does-not-exist")
except TagLoggerError as err:
    err.code     # stable machine-readable code, e.g. "not_found"
    err.status   # HTTP status, or 0 for a transport/network failure
    err.message
    err.request_id          # from X-Request-Id, when present
    err.is_rate_limit       # True on HTTP 429
    err.is_transport        # True when the request never reached the server
    err.retry_after_seconds # parsed from Retry-After, when present

Analytics — taglogger_sdk.analytics

A bundled, dependency-free toolkit that turns the location data the API returns into stops, trips, dwell intervals, clusters and map-ready shapes. Everything runs locally; nothing is sent anywhere.

from taglogger_sdk.analytics import (
    fixes_from_location_points,
    condition_fixes,
    detect_stops,
    detect_trips,
    to_feature_collection,
)

# Pull a tag's history, then analyze it in-process
page = client.tags.history("tag-1", order="asc", limit=500)
fixes = fixes_from_location_points(page["data"])

clean = condition_fixes(fixes, max_accuracy_meters=100)
stops = detect_stops(clean, radius_meters=50, min_duration_ms=300_000)  # 5 min
trips = detect_trips(clean, stops)

geojson = to_feature_collection(trips=trips, stops=stops)  # ready for any map

Battery projection from a series of fractional-level readings (0.0–1.0) that you collect over time. Note that client.tags.battery("tag-1") reports a coarse integer status code (0 unknown … 4 critical; see BatteryStatus) and its numeric level is currently always None, so the estimator is designed for series you assemble yourself — for example by recording the last known good fractional level alongside its timestamp:

from taglogger_sdk.analytics import BatterySample, estimate_battery_days_remaining

# Two fractional-level samples taken roughly a week apart.
est = estimate_battery_days_remaining([
    BatterySample(level=0.92, timestamp_ms=1_700_000_000_000),
    BatterySample(level=0.58, timestamp_ms=1_700_600_000_000),
])
print(est.days_remaining, est.drain_per_day, est.reason)

Available helpers

• Types & adapters: Fix, Stop, Trip, Cluster, DwellInterval, BoundingBox, LatLng, BatterySample, BatteryEstimate, fix_from_location_point, fixes_from_location_points
• Conditioning: sort_by_time, dedupe_by_timestamp, condition_fixes
• Segmentation: detect_stops, detect_trips, compute_dwell, cluster_fixes (DBSCAN)
• Geometry: haversine_meters, bearing_degrees, centroid, bounding_box, path_length_meters, point_in_polygon, is_within_radius, douglas_peucker, encode_polyline, decode_polyline
• Shaping: to_point, to_line_string, to_polyline, to_feature_collection, trip_to_feature, stop_to_feature, fixes_to_feature

Performance: douglas_peucker runs in O(n log n) on typical tracks but can reach O(n**2) on a path that retains nearly every vertex (a dense zig-zag). For very large inputs, down-sample first or raise epsilon_meters. encode_polyline raises ValueError on non-finite (nan/inf) coordinates.

Custom transport

Networking is pluggable. Inject any callable matching the transport signature to run hermetic tests or route requests through your own stack:

from taglogger_sdk import TagLoggerClient
from taglogger_sdk.http import HttpResponse

def fake_transport(method, url, headers, body, timeout):
    return HttpResponse(status=200, headers={"content-type": "application/json"}, text='{"data": []}')

client = TagLoggerClient(api_key="tl_test_x", transport=fake_transport)

License

MIT