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:
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://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())
PickupEvent
ObjectThe PickupEvent
object that is returned from the above calls comes with three
properties:
date
: a datetime.date
that denotes the pickup datepickup_types
: a list of PickupType
objects that will occur with this eventarea_name
: the name of the area in which the event is occurringPickupType
ObjectThe 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!
python3 -m venv .venv
source ./.venv/bin/activate
script/setup
poetry run pytest --cov aiorecollect tests
README.md
with any new documentation.