edward/agenda
1
0
Fork 0
Code Issues 16 Pull requests Packages Projects Releases Wiki Activity

Add refresh option for Bristol waste schedule

Browse source
This commit is contained in:
Edward Betts 2024-01-08 15:45:08 +00:00
parent 7456f72325
commit 199eb82bce
1 changed files with 8 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
agenda/waste_schedule.py
View file

@ -104,7 +104,9 @@ def parse(root: lxml.html.HtmlElement) -> list[Event]:
BristolSchedule = list[dict[str, typing.Any]] BristolSchedule = list[dict[str, typing.Any]]
async def get_bristol_data(data_dir: str, uprn: str) -> BristolSchedule: async def get_bristol_data(
data_dir: str, uprn: str, refresh: bool = False
) -> BristolSchedule:
"""Get Bristol Waste schedule, with cache.""" """Get Bristol Waste schedule, with cache."""
now = datetime.now() now = datetime.now()
waste_dir = os.path.join(data_dir, "waste") waste_dir = os.path.join(data_dir, "waste")
@ -122,7 +124,7 @@ async def get_bristol_data(data_dir: str, uprn: str) -> BristolSchedule:
json_data = json.load(open(os.path.join(waste_dir, recent_filename))) json_data = json.load(open(os.path.join(waste_dir, recent_filename)))
return typing.cast(BristolSchedule, json_data["data"]) return typing.cast(BristolSchedule, json_data["data"])
if existing and delta < timedelta(hours=ttl_hours): if not refresh and existing and delta < timedelta(hours=ttl_hours):
return get_from_recent() return get_from_recent()
try: try:
@ -186,9 +188,11 @@ async def get_bristol_gov_uk_data(uprn: str) -> httpx.Response:
return response return response
async def get_bristol_gov_uk(start_date: date, data_dir: str, uprn: str) -> list[Event]: async def get_bristol_gov_uk(
start_date: date, data_dir: str, uprn: str, refresh: bool = False
) -> list[Event]:
"""Get waste collection schedule from Bristol City Council.""" """Get waste collection schedule from Bristol City Council."""
data = await get_bristol_data(data_dir, uprn) data = await get_bristol_data(data_dir, uprn, refresh)
by_date: defaultdict[date, list[str]] = defaultdict(list) by_date: defaultdict[date, list[str]] = defaultdict(list)