aiopurpleair

Python client library for the PurpleAir air-quality sensors 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.0:

Summary:

• 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.

⚠️ Breaking Changes:

• API-key check moved from api.async_check_api_key() to api.keys.async_check_api_key(), for 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())
   

See Usage for detailed usage instructions.

Table of Contents

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

Features

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.

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())

Installation

Project integration:

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

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

The distribution name is ptr727-aiopurpleair. aiopurpleair supports Python 3.13 and 3.14, and depends on aiohttp, pydantic, yarl, and certifi.

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.

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 version.json (1.0 base) 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 push to main (stable) or develop (prerelease), or a manual dispatch, cuts a GitHub Release and uploads the wheel + sdist to PyPI. 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 AGENTS.md for how the code is validated against the spec.

Coverage: all 11 paths of the spec (currently API 1.2.0) 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.

Contributing

• Branching workflow:
  • The repo uses a two-branch model with ruleset-enforced merge methods.
  • Feature branch -> develop via squash merge (develop is kept linear).
  • develop -> main via merge commit (preserves develop's commit list on main as the second parent of each release commit).
  • Dependabot targets main and develop in parallel via separate PRs and auto-merges every tier once the required check passes.
  • See WORKFLOW.md and AGENTS.md for complete details.
• Code style:
  • See CODESTYLE.md and .editorconfig for Python code style rules. Everything runs through uv run (ruff, mypy, pyright, pytest with 100% coverage and syrupy snapshots).
• Repository setup:
  • See repo-config/README.md for repo configuration details.

Origin

aiopurpleair is an independent, MIT-licensed continuation of the bachya/aiopurpleair PurpleAir API client. Its distinguishing capabilities originated in two upstream contributions that were abandoned:

• A pull request against bachya/aiopurpleair adding the organization endpoint and typed error codes (bachya/aiopurpleair#719), since abandoned.
• A PurpleAir integration proposed for Home Assistant core as home-assistant/core#140901, now maintained separately as the homeassistant-purpleair HACS custom integration, this library's primary consumer.

The import package name is aiopurpleair; the distribution name ptr727-aiopurpleair keeps it distinct from the canonical aiopurpleair on PyPI. The original MIT copyright is retained alongside the current maintainer's in LICENSE and NOTICE.

License

Licensed under the MIT License

• Current maintainer: Pieter Viljoen (@ptr727).
• Original aiopurpleair author: Aaron Bach (@bachya).