aiorecollect
is a Python 3, asyncio-based library for the ReCollect Waste API. It allows
users to programmatically retrieve schedules for waste removal in their area, including
trash, recycling, compost, and more.
Special thanks to @stealthhacker for the inspiration!
pip install aiorecollect
aiorecollect
is currently supported on:
- Python 3.10
- Python 3.11
- Python 3.12
To use aiorecollect
, you must know both your ReCollect Place and Service IDs.
In general, cities/municipalities that utilize ReCollect will give you a way to subscribe to a calendar with pickup dates. If you examine the iCal URL for this calendar, the Place and Service IDs are embedded in it:
webcal:https://recollect.a.ssl.fastly.net/api/places/PLACE_ID/services/SERVICE_ID/events.en-US.ics
import asyncio
from datetime import date
from aiorecollect import Client
async def main() -> None:
"""Run."""
client = await Client("<PLACE ID>", "<SERVICE ID>")
# The client has a few attributes that you can access:
client.place_id
client.service_id
# Get all pickup events on the calendar:
pickup_events = await client.async_get_pickup_events()
# ...or get all pickup events within a certain date range:
pickup_events = await client.async_get_pickup_events(
start_date=date(2020, 10, 1), end_date=date(2020, 10, 31)
)
# ...or just get the next pickup event:
next_pickup = await client.async_get_next_pickup_event()
asyncio.run(main())
The PickupEvent
object that is returned from the above calls comes with three
properties:
date
: adatetime.date
that denotes the pickup datepickup_types
: a list ofPickupType
objects that will occur with this eventarea_name
: the name of the area in which the event is occurring
The PickupType
object contains the "internal" name of the pickup type and a
human-friendly representation when it exists:
name
: the internal name of the pickup typefriendly_name
: the humany-friendly name of the pickup type (if it exists)
By default, the library creates a new connection to ReCollect 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 aiorecollect import Client
async def main() -> None:
"""Run."""
async with ClientSession() as session:
client = await Client("<PLACE ID>", "<SERVICE ID>", session=session)
# Get to work...
asyncio.run(main())
Thanks to all of our contributors so far!
- Check for open features/bugs or initiate a discussion on one.
- Fork the repository.
- (optional, but highly recommended) Create a virtual environment:
python3 -m venv .venv
- (optional, but highly recommended) Enter the virtual environment:
source ./.venv/bin/activate
- Install the dev environment:
script/setup
- Code your new feature or bug fix on a new branch.
- Write tests that cover your new functionality.
- Run tests and ensure 100% code coverage:
poetry run pytest --cov aiorecollect tests
- Update
README.md
with any new documentation. - Submit a pull request!