This Python library is a tiny OpenADR3 library that can be used to perform a minimal set of operations towards a VTN.
Currently, it supports the following operations:
- List events [GET]
- Filter by program id
- Filter by target type and target values
- Limit and skip for pagination
- List programs [GET]
- Filter by target type and target values
- Limit and skip for pagination
- List reports [GET]
- Filter by program id
- Filter by event id
- Filter by client name
- Limit and skip for pagination
- List subscriptions [GET]
- Filter by program id
- Filter by client name
- Filter by target type and target values
- Filter by objects
- Limit and skip for pagination
- Update a subscription [PUT]
- Delete a subscription [DELETE]
- Get a subscription by id [GET]
- Update a program [PUT]
- Delete a program [DELETE]
- Get a program by id [GET]
- Create a report [POST]
- Create a subscription [POST]
- Create a report object based on an initial event
A small example of how to list programs and events and create a report:
import asyncio
import toadr3
OAUTH_CONFIG = toadr3.OAuthConfig(
token_url="", # URL to the OAuth2 token endpoint
grant_type="", # OAuth2 grant type
claims={"scope": ""}, # OAuth2 claims, for example 'scope'
client_id="", # OAuth2 client ID or set to None use environment variable
client_secret="", # OAuth2 secret or set to None use environment variable
)
VTN_URL = "" # URL to the VTN
async def main():
async with toadr3.ToadrClient(vtn_url=VTN_URL, oauth_config=OAUTH_CONFIG) as client:
programs = await client.get_programs()
for program in programs:
print(f"Program: ID={program.id}, Name={program.program_name}")
events = await client.get_events()
for event in events:
print(f"Event: ID={event.id}, Name={event.event_name}, Date={event.created_date_time}")
report = toadr3.models.Report.create_report(
event=events[0],
client_name="ReadmeClient",
report_type="README_REPORT",
report_values=[True],
)
try:
result = await client.post_report(report)
print(f"Report created with ID={result.id}")
except toadr3.ToadrError as e:
print(f"ToadrError: {e}")
if __name__ == '__main__':
asyncio.run(main())