Make thespacedevs async

This commit is contained in:
Edward Betts 2023-11-05 14:47:17 +00:00
parent 35ec8008a8
commit 997af2fde6
2 changed files with 10 additions and 8 deletions

View file

@ -21,8 +21,6 @@ import yaml
from dateutil.easter import easter
from dateutil.relativedelta import FR, relativedelta
from agenda import thespacedevs
from . import (
birthday,
calendar,
@ -32,6 +30,7 @@ from . import (
gwr,
markets,
sun,
thespacedevs,
travel,
waste_schedule,
)
@ -301,10 +300,12 @@ async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
gbpusd,
gwr_advance_tickets,
bank_holiday,
rockets,
) = await asyncio.gather(
fx.get_gbpusd(config),
gwr.advance_ticket_date(data_dir),
get_next_bank_holiday(today),
thespacedevs.get_launches(rocket_dir, limit=40),
)
reply = {
@ -328,7 +329,7 @@ async def get_data(now: datetime) -> typing.Mapping[str, str | object]:
+ markets.tobacco_factory(last_year)
+ markets.nailsea_farmers(last_year)
),
"rockets": thespacedevs.get_launches(rocket_dir, limit=40),
"rockets": rockets,
}
skip = {"now", "gbpusd", "rockets", "stock_markets", "xmas_last_posting_dates"}

View file

@ -3,20 +3,21 @@ import os
import typing
from datetime import datetime
import requests
import httpx
Launch = dict[str, typing.Any]
Summary = dict[str, typing.Any]
def next_launch_api(rocket_dir: str, limit: int = 200) -> list[Launch]:
async def next_launch_api(rocket_dir: str, limit: int = 200) -> list[Launch]:
"""Get the next upcoming launches from the API."""
now = datetime.now()
filename = os.path.join(rocket_dir, now.strftime("%Y-%m-%d_%H:%M:%S.json"))
url = "https://ll.thespacedevs.com/2.2.0/launch/upcoming/"
params: dict[str, str | int] = {"limit": limit}
r = requests.get(url, params=params)
async with httpx.AsyncClient() as client:
r = await client.get(url, params=params)
open(filename, "w").write(r.text)
data = r.json()
return [summarize_launch(launch) for launch in data["results"]]
@ -107,7 +108,7 @@ def summarize_launch(launch: Launch) -> Summary:
}
def get_launches(rocket_dir: str, limit: int = 200) -> list[Summary]:
async def get_launches(rocket_dir: str, limit: int = 200) -> list[Summary]:
"""Get rocket launches with caching."""
now = datetime.now()
existing = [x for x in (filename_timestamp(f) for f in os.listdir(rocket_dir)) if x]
@ -116,7 +117,7 @@ def get_launches(rocket_dir: str, limit: int = 200) -> list[Summary]:
if not existing or (now - existing[0][0]).seconds > 3600: # one hour
try:
return next_launch_api(rocket_dir, limit=limit)
return await next_launch_api(rocket_dir, limit=limit)
except ValueError:
print("*** SpaceX next launch error ***")