Don't bother with httpx for the space launch API
This commit is contained in:
		
							parent
							
								
									073f452356
								
							
						
					
					
						commit
						566b09f888
					
				| 
						 | 
				
			
			@ -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]
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -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)
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
		Loading…
	
		Reference in a new issue