aiopurpleair-ptr727 2026.6.0

Fork of aiopurpleair with extra exception subclasses and the GET /v1/organization endpoint.

🟣 aiopurpleair: A Python3, asyncio-based library to interact with the PurpleAir API

aiopurpleair is a Python3, asyncio-based library to interact with the PurpleAir API.

• Installation
• Python Versions
• Usage
  • Checking an API Key
  • Getting Sensors
  • Getting a Single Sensor
  • Getting a Private Sensor
  • Getting Nearby Sensors
  • Getting a Map URL
  • Connection Pooling
• Contributing

Installation

pip install aiopurpleair

Python Versions

aiopurpleair is currently supported on:

• Python 3.10
• Python 3.11
• Python 3.12

Usage

In-depth documentation on the API can be found here. Unless otherwise noted, aiopurpleair endeavors to follow the API as closely as possible.

Checking an API Key

To check whether an API key is valid and what properties it has:

import asyncio

from aiopurpleair import API


async def main() -> None:
    """Run."""
    api = API("<API KEY>")
    response = await api.async_check_api_key()
    # >>> response.api_key_type == ApiKeyType.READ
    # >>> response.api_version == "V1.0.11-0.0.41"
    # >>> response.timestamp_utc == datetime(2022, 10, 27, 18, 25, 41)


asyncio.run(main())

Getting Sensors

import asyncio

from aiopurpleair import API


async def main() -> None:
    """Run."""
    api = API("<API_KEY>")
    response = await api.sensors.async_get_sensors(["name"])
    # >>> response.api_version == "V1.0.11-0.0.41"
    # >>> response.data == {
    # >>>     131075: SensorModel(sensor_index=131075, name=Mariners Bluff),
    # >>>     131079: SensorModel(sensor_index=131079, name=BRSKBV-outside),
    # >>> }
    # >>> response.data_timestamp_utc == datetime(2022, 11, 3, 19, 25, 31)
    # >>> response.fields == ["sensor_index", "name"]
    # >>> response.firmware_default_version == "7.02"
    # >>> response.max_age == 604800
    # >>> response.timestamp_utc == datetime(2022, 11, 3, 19, 26, 29)


asyncio.run(main())

Method Parameters

• fields (required): The sensor data fields to include
• location_type (optional): An LocationType to filter by
• max_age (optional): Filter results modified within these seconds
• modified_since (optional): Filter results modified since a UTC datetime
• read_keys (optional): Read keys for private sensors
• sensor_indices (optional): Filter results by sensor index

Getting a Single Sensor

import asyncio

from aiopurpleair import API


async def main() -> None:
    """Run."""
    api = API("<API_KEY>")
    response = await api.sensors.async_get_sensor(131075)
    # >>> response.api_version == "V1.0.11-0.0.41"
    # >>> response.data_timestamp_utc == datetime(2022, 11, 5, 16, 36, 21)
    # >>> response.sensor == SensorModel(sensor_index=131075, ...),
    # >>> response.timestamp_utc == datetime(2022, 11, 5, 16, 37, 3)


asyncio.run(main())

Method Parameters

• sensor_index (required): The sensor index of the sensor to retrieve.
• fields (optional): The sensor data fields to include.
• read_key (optional): A read key for a private sensor.

Getting a Private Sensor

Sensors registered as private require their per-sensor read key (visible to the sensor owner from the PurpleAir dashboard). Without it, the API returns NotFoundError even though the sensor exists. Pass the key as read_key= for a single sensor, or as a list to read_keys= when querying multiple sensors at once (the library joins the list with commas before sending):

import asyncio

from aiopurpleair import API


async def main() -> None:
    """Run."""
    api = API("<API_KEY>")

    # Single private sensor
    response = await api.sensors.async_get_sensor(131075, read_key="<SENSOR_READ_KEY>")
    # >>> response.sensor == SensorModel(sensor_index=131075, ...)

    # Multiple sensors, mixing public and private — only the private ones need a key
    response = await api.sensors.async_get_sensors(
        ["name", "pm2.5"],
        sensor_indices=[131075, 131079],
        read_keys=["<SENSOR_READ_KEY_FOR_131075>"],
    )
    # >>> response.data == {131075: ..., 131079: ...}


asyncio.run(main())

Getting Nearby Sensors

This method returns a list of NearbySensorResult objects that are within a bounding box around a given latitude/longitude pair. The list is sorted from nearest to furthest (i.e., the first index in the list is the closest to the latitude/longitude).

NearbySensorResult objects have two properties:

• sensor: the corresponding SensorModel object
• distance: the calculated distance (in kilometers) between this sensor and the provided latitude/longitude

import asyncio

from aiopurpleair import API


async def main() -> None:
    """Run."""
    api = API("<API_KEY>")
    sensors = await api.sensors.async_get_nearby_sensors(
        ["name"], 51.5285582, -0.2416796, 10
    )
    # >>> [NearbySensorResult(...), NearbySensorResult(...)]


asyncio.run(main())

Method Parameters

• fields (required): The sensor data fields to include
• latitude (required): The latitude of the point to measure distance from
• longitude (required): The longitude of the point to measure distance from
• distance (required): The distance from the measured point to search (in kilometers)
• limit (optional): Limit the results

Getting a Map URL

If you need to get the URL to a particular sensor index on the PurpleAir map website, simply pass the appropriate sensor index to the get_map_url method:

import asyncio

from aiopurpleair import API


async def main() -> None:
    """Run."""
    api = API("<API_KEY>")
    map_url = api.get_map_url(12345)
    # >>> https://map.purpleair.com/1/mAQI/a10/p604800/cC0?select=12345


asyncio.run(main())

Connection Pooling

By default, the library creates a new connection to the PurpleAir API with each coroutine. If you are calling a large number of coroutines (or merely want to squeeze out every second of runtime savings possible), an aiohttp ClientSession can be used for connection pooling:

import asyncio

from aiohttp import ClientSession

from aiopurpleair import API


async def main() -> None:
    """Run."""
    async with ClientSession() as session:
        api = API("<API KEY>", session=session)

        # Get to work...


asyncio.run(main())

Contributing

Thanks to all of our contributors so far!

1. Check for open features/bugs or initiate a discussion on one.
2. Fork the repository.
3. (optional, but highly recommended) Create a virtual environment: python3 -m venv .venv
4. (optional, but highly recommended) Enter the virtual environment: source ./.venv/bin/activate
5. Install the dev environment: script/setup
6. Code your new feature or bug fix on a new branch.
7. Write tests that cover your new functionality.
8. Run tests and ensure 100% code coverage: poetry run pytest --cov aiopurpleair tests
9. Update README.md with any new documentation.
10. Submit a pull request!