parent
cb2a2c7fb8
commit
e0735b4185
|
@ -10,6 +10,8 @@ import httpx
|
||||||
Launch = dict[str, typing.Any]
|
Launch = dict[str, typing.Any]
|
||||||
Summary = dict[str, typing.Any]
|
Summary = dict[str, typing.Any]
|
||||||
|
|
||||||
|
ttl = 60 * 60 * 2 # two hours
|
||||||
|
|
||||||
|
|
||||||
async 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."""
|
"""Get the next upcoming launches from the API."""
|
||||||
|
@ -137,14 +139,16 @@ def summarize_launch(launch: Launch) -> Summary:
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
async def get_launches(rocket_dir: str, limit: int = 200) -> list[Summary]:
|
async def get_launches(
|
||||||
|
rocket_dir: str, limit: int = 200, refresh: bool = False
|
||||||
|
) -> list[Summary]:
|
||||||
"""Get rocket launches with caching."""
|
"""Get rocket launches with caching."""
|
||||||
now = datetime.now()
|
now = datetime.now()
|
||||||
existing = [x for x in (filename_timestamp(f) for f in os.listdir(rocket_dir)) if x]
|
existing = [x for x in (filename_timestamp(f) for f in os.listdir(rocket_dir)) if x]
|
||||||
|
|
||||||
existing.sort(reverse=True)
|
existing.sort(reverse=True)
|
||||||
|
|
||||||
if not existing or (now - existing[0][0]).seconds > 3600: # one hour
|
if refresh or not existing or (now - existing[0][0]).seconds > ttl:
|
||||||
try:
|
try:
|
||||||
return await next_launch_api(rocket_dir, limit=limit)
|
return await next_launch_api(rocket_dir, limit=limit)
|
||||||
except httpx.ReadTimeout:
|
except httpx.ReadTimeout:
|
||||||
|
|
26
update_thespacedevs.py
Executable file
26
update_thespacedevs.py
Executable file
|
@ -0,0 +1,26 @@
|
||||||
|
#!/usr/bin/python3
|
||||||
|
"""Update cache of space launch API."""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
from time import time
|
||||||
|
|
||||||
|
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.get_launches(rocket_dir, limit=200, refresh=True)
|
||||||
|
|
||||||
|
|
||||||
|
t0 = time()
|
||||||
|
rockets = asyncio.run(get_launches())
|
||||||
|
time_taken = time() - t0
|
||||||
|
if not sys.stdin.isatty():
|
||||||
|
sys.exit(0)
|
||||||
|
print(len(rockets), "launches")
|
||||||
|
print(f"took {time_taken:.1f} seconds")
|
Loading…
Reference in a new issue