diff --git a/agenda/thespacedevs.py b/agenda/thespacedevs.py index c1c785e..130be2f 100644 --- a/agenda/thespacedevs.py +++ b/agenda/thespacedevs.py @@ -5,7 +5,7 @@ import os import typing from datetime import datetime -import httpx +import requests Launch = dict[str, typing.Any] Summary = dict[str, typing.Any] @@ -13,15 +13,14 @@ Summary = dict[str, typing.Any] ttl = 60 * 60 * 2 # two hours -async def next_launch_api(rocket_dir: str, limit: int = 200) -> list[Launch]: +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} - async with httpx.AsyncClient() as client: - r = await client.get(url, params=params) + r = requests.get(url, params=params) open(filename, "w").write(r.text) data = r.json() return [summarize_launch(launch) for launch in data["results"]] @@ -151,7 +150,7 @@ def read_cached_launches(rocket_dir: str) -> list[Summary]: return [summarize_launch(launch) for launch in data["results"]] -async def get_launches( +def get_launches( rocket_dir: str, limit: int = 200, refresh: bool = False ) -> list[Summary]: """Get rocket launches with caching.""" @@ -161,10 +160,7 @@ async def get_launches( existing.sort(reverse=True) if refresh or not existing or (now - existing[0][0]).seconds > ttl: - try: - return await next_launch_api(rocket_dir, limit=limit) - except httpx.ReadTimeout: - pass + return next_launch_api(rocket_dir, limit=limit) f = existing[0][1] diff --git a/update_thespacedevs.py b/update_thespacedevs.py index 44a96f8..e5b9ec8 100755 --- a/update_thespacedevs.py +++ b/update_thespacedevs.py @@ -1,7 +1,6 @@ #!/usr/bin/python3 """Update cache of space launch API.""" -import asyncio import os import sys from time import time @@ -11,14 +10,9 @@ import agenda.thespacedevs config = __import__("config.default", fromlist=[""]) rocket_dir = os.path.join(config.DATA_DIR, "thespacedevs") - -async def get_launches() -> list[agenda.thespacedevs.Launch]: - """Call space launch API and cache results.""" - return await agenda.thespacedevs.next_launch_api(rocket_dir) - - t0 = time() -rockets = asyncio.run(get_launches()) + +rockets = agenda.thespacedevs.next_launch_api(rocket_dir) time_taken = time() - t0 if not sys.stdin.isatty(): sys.exit(0)