aiopurpleair

Async Python client library for the PurpleAir air-quality sensor API.

Build and Distribution

• Source Code: GitHub - Source code, issues, discussions, and CI/CD pipelines.
• Releases: GitHub Releases - Version tagged source code and build artifacts.
• PyPI Packages: PyPI - Python library published to PyPI.org as ptr727-aiopurpleair.

Build Status

Releases

Release Notes

Version 1.0:

• Initial release of the library published as ptr727-aiopurpleair, and the continuation of the abandoned upstream PR bachya/aiopurpleair#719.
• New endpoints added:
  • Organization (GET /v1/organization) to get remaining API points and consumption rate.
  • Sensor history (GET /v1/sensors/:sensor_index/history[/csv]) as JSON or CSV.
  • Groups (/v1/groups*) for group and member management.
• Typed exceptions for every documented PurpleAir error code.
• Reconstructed OpenAPI spec from the upstream apiDoc data with an automated script.
• Typed timezone-aware models parse to explicit-UTC datetime objects.
• Updated packaging using hatchling and uv, automatic versioning using NBGV, PyPI OIDC Trusted-Publishing releases, a 100% coverage gate, and syrupy snapshot tests.
• ⚠️ API-key check moved from api.async_check_api_key() to api.keys.async_check_api_key(). Maintains association consistency alongside api.sensors, api.organizations, and api.groups.

See Release History for complete release notes and older versions.

Getting Started

Get started with aiopurpleair in two easy steps:

1. Add aiopurpleair to your project:

   # Add the package to your project (import name stays `aiopurpleair`)
   pip install ptr727-aiopurpleair
   

2. Write some code:

   import asyncio
   
   from aiopurpleair import API
   
   
   async def main() -> None:
       """Check an API key and fetch sensors."""
       api = API("<API_KEY>")
       keys = await api.keys.async_check_api_key()
       sensors = await api.sensors.async_get_sensors(["name", "pm2.5"])
       organization = await api.organizations.async_get_organization()
   
   
   asyncio.run(main())
   

Table of Contents

• aiopurpleair
  • Build and Distribution
    • Build Status
    • Releases
    • Release Notes
  • Getting Started
  • Table of Contents
  • Overview
  • Installation
  • Usage
    • Checking an API Key
    • Getting Sensors
    • Getting Sensor History
    • Getting the Organization
    • Working with Groups
    • Error Handling
    • Connection Pooling
  • Build Artifacts
  • API Reference
  • Questions or Issues
  • Contributing
  • 3rd Party Tools
  • Credits
  • License

Overview

Full async coverage of the PurpleAir API, each method mirroring a documented endpoint:

• Keys - validate an API key and read its type (GET /v1/keys), via api.keys.async_check_api_key().
• Sensors - one sensor, many sensors by field selection, or a distance-sorted nearby search, plus a map-URL helper (GET /v1/sensors, GET /v1/sensors/{sensor_index}), via api.sensors.
• Sensor history - historical time series for a sensor as parsed JSON or raw CSV (GET /v1/sensors/{sensor_index}/history[/csv]), via api.sensors.
• Organization - the account's remaining API points and consumption rate (GET /v1/organization), via api.organizations.
• Groups - create, list, inspect, and delete groups; add and remove member sensors; read member sensor data and member history CSV (/v1/groups*), via api.groups.
• Typed errors - each documented API error code maps to a specific PurpleAirError subclass, so callers catch a precise condition instead of parsing str(err).
• Timezone-aware UTC datetimes and typed Pydantic response models, shipped with a py.typed marker.
• Modern packaging: hatchling, uv, automatic versioning, OIDC-published releases, and 100% test coverage.

Installation

Project integration:

# Add the package to your project
pip install ptr727-aiopurpleair

# Import the library (the import name stays `aiopurpleair`)
import aiopurpleair

Dependencies:

Requires Python 3.13 or later, and depends on aiohttp, pydantic, yarl, and certifi.

Usage

In-depth documentation on the API is available from PurpleAir. Unless otherwise noted, aiopurpleair follows the API as closely as possible.

Checking an API Key

import asyncio

from aiopurpleair import API


async def main() -> None:
    """Check whether an API key is valid and what properties it has."""
    api = API("<API_KEY>")
    response = await api.keys.async_check_api_key()
    # >>> response.api_key_type == ApiKeyType.READ
    # >>> response.api_version == "V1.0.11-0.0.41"


asyncio.run(main())

Getting Sensors

import asyncio

from aiopurpleair import API


async def main() -> None:
    """Fetch sensor data for the requested fields."""
    api = API("<API_KEY>")
    response = await api.sensors.async_get_sensors(["name", "pm2.5"])
    # >>> response.data == {131075: SensorModel(...), 131079: SensorModel(...)}


asyncio.run(main())

Private sensors require their per-sensor read key: pass read_key= to async_get_sensor, or read_keys=[...] to async_get_sensors. Use async_get_nearby_sensors(fields, latitude, longitude, distance) for a distance-sorted search, and get_map_url(sensor_index) for a map link.

Getting Sensor History

Fetch a historical time series for a sensor, as parsed JSON or as raw CSV. The averaging period is in minutes (e.g. 0 for real-time, 60 for hourly, 1440 for daily):

import asyncio
from datetime import UTC, datetime, timedelta

from aiopurpleair import API


async def main() -> None:
    """Fetch a day of hourly history for a sensor."""
    api = API("<API_KEY>")
    end = datetime.now(UTC)
    start = end - timedelta(days=1)

    history = await api.sensors.async_get_sensor_history(
        131075,
        ["humidity", "temperature", "pm2.5_atm"],
        start_timestamp_utc=start,
        end_timestamp_utc=end,
        average=60,
    )
    # >>> history.data == [{"time_stamp": 1667336400, "humidity": 37, ...}, ...]

    csv = await api.sensors.async_get_sensor_history_csv(
        131075, ["pm2.5_atm"], start_timestamp_utc=start, end_timestamp_utc=end, average=60
    )
    # >>> csv.startswith("time_stamp,sensor_index,pm2.5_atm")


asyncio.run(main())

The history endpoint is a gated feature; if it is not enabled for your API key the call raises ApiDisabledError.

Getting the Organization

The organization endpoint reports the account's remaining API points and consumption rate, useful for surfacing a low-points warning before queries start failing:

import asyncio

from aiopurpleair import API


async def main() -> None:
    """Fetch the organization associated with the API key."""
    api = API("<API_KEY>")
    response = await api.organizations.async_get_organization()
    # >>> response.remaining_points == 500000
    # >>> response.consumption_rate == 1234.5
    # >>> response.organization_id == "..."
    # >>> response.organization_name == "..."


asyncio.run(main())

Working with Groups

Groups organize sensors for data access. Create and delete operations require a WRITE key; reads (list, detail, member data) use a READ key:

import asyncio

from aiopurpleair import API


async def main() -> None:
    """Create a group, add a member, read it back, then clean up."""
    write_api = API("<WRITE_API_KEY>")
    read_api = API("<READ_API_KEY>")

    created = await write_api.groups.async_create_group("My Sensors")
    group_id = created.group_id

    await write_api.groups.async_create_member(group_id, sensor_index=131075)

    groups = await read_api.groups.async_get_groups()
    detail = await read_api.groups.async_get_group(group_id)
    # >>> detail.members == [GroupMember(id=..., sensor_index=131075, ...)]

    members = await read_api.groups.async_get_members(group_id, ["name", "pm2.5"])
    # >>> members.data == {131075: SensorModel(...)}

    await write_api.groups.async_delete_group(group_id)


asyncio.run(main())

Adding a private sensor also requires its registration owner_email. Per-member history is available as CSV via async_get_member_history_csv(group_id, member_id, fields, ...).

Error Handling

Each documented PurpleAir API error code maps to a specific exception subclass, so callers can catch a precise condition instead of pattern-matching on str(err). Every subclass derives from PurpleAirError:

import asyncio

from aiopurpleair import API
from aiopurpleair.errors import InvalidApiKeyError, RateLimitExceededError


async def main() -> None:
    """Handle specific PurpleAir error conditions."""
    api = API("<API_KEY>")
    try:
        await api.sensors.async_get_sensors(["name"])
    except InvalidApiKeyError:
        ...  # the API key is missing or invalid
    except RateLimitExceededError:
        ...  # back off and retry later


asyncio.run(main())

All error codes and semantics are verified against the official PurpleAir API documentation.

Connection Pooling

By default a new connection is created per coroutine. Pass an existing aiohttp ClientSession for connection pooling:

import asyncio

from aiohttp import ClientSession

from aiopurpleair import API


async def main() -> None:
    """Reuse a session across calls."""
    async with ClientSession() as session:
        api = API("<API_KEY>", session=session)
        ...


asyncio.run(main())

Build Artifacts

Build process and artifacts:

• Package: a Python wheel + sdist (ptr727-aiopurpleair), built with the hatchling backend on a src-layout (src/aiopurpleair/) and managed with uv.
• Versioning: automatic via Nerdbank.GitVersioning from the major.minor base in version.json plus git height. main builds a clean stable X.Y.Z, develop a X.Y.Z.dev0 prerelease. There is no manual tagging.
• Publishing: releases publish to PyPI over OIDC Trusted Publishing (no stored API token). A shipped-path merge to main by an allowlisted bot (stable), or a manual dispatch of main (stable) or develop (prerelease), cuts a GitHub Release and uploads the wheel + sdist to PyPI. A human merge never auto-cuts a release. See WORKFLOW.md for the complete CI/CD contract.

API Reference

PurpleAir does not publish an OpenAPI/Swagger spec. This repo reconstructs one at docs/purpleair-openapi.yaml from PurpleAir's apiDoc-generated docs (which serve machine-readable api_data.js), using scripts/generate_openapi.py. The library's endpoint, field, and error-code coverage is validated against this spec.

Regenerate it after an upstream API change:

# Live-fetch https://api.purpleair.com/api_data.js, rebuild and validate the spec
uv run --with pyyaml --with openapi-spec-validator python scripts/generate_openapi.py

The generator takes the API version from the docs' changelog (the apiDoc build-metadata version lags behind), validates the result, and writes docs/purpleair-openapi.yaml. A non-empty diff means the upstream API changed. See ARCHITECTURE.md for how the code is validated against the spec.

Coverage: all 11 paths of the reconstructed spec (docs/purpleair-openapi.yaml) are implemented: keys, sensors (list, single, and history JSON/CSV), organization, and the full Groups API (group and member management, member data, and member history). The single-sensor stats/stats_a/stats_b blocks are returned as part of the sensor payload but are not requestable fields values, so they are parsed on the response but excluded from the requestable field catalog.

Questions or Issues

• General questions:
  • Use the Discussions forum for general questions.
• Bug reports:
  • Ask in the Discussions forum if you are not sure if it is a bug.
  • Check the existing Issues tracker for known problems.
  • If the issue is unique and a bug, file it in Issues, and include all pertinent steps to reproduce the issue.

Contributing

• Branching workflow:
  • Feature branch -> develop via squash merge; develop -> main via merge commit. Both methods are pinned in the branch rulesets.
  • CI runs on every pull request. A fork pull request runs the base-repo check and can satisfy it, though a first-time contributor's run waits on a maintainer's approval, per the repository's first_time_contributors policy.
  • Dependabot targets main and develop in parallel and auto-merges once the required check passes.
  • See WORKFLOW.md for the CI/CD contract and OPERATIONS.md for the release runbooks.
• Code style:
  • ruff, mypy, and pyright; see CODESTYLE.md and .editorconfig. Everything runs through uv run (with pytest at 100% coverage and syrupy snapshots).
• Repository setup:
  • Settings, labels, and branch rulesets are applied and audited by a hub-hosted script rather than one carried here. See OPERATIONS.md "Configuration Layout" for what this repository keeps and AUDIT.md section 4 for how to run the check.

3rd Party Tools

The third-party tools, libraries, and actions this project depends on.

Credits

This library is an independent implementation based on the bachya/aiopurpleair PurpleAir API client by Aaron Bach (@bachya).
It was created to be maintained independently after the upstream PR bachya/aiopurpleair#719 - adding organization support - was abandoned.

The original MIT copyright is retained alongside that of the current maintainer in LICENSE and NOTICE.

License

Licensed under the MIT License and NOTICE