import httpx # requests capability, but can work with async
from prefect import flow, task, logging
from datetime import timedelta
@task(
retries=3,
# cache_key_fn=
persist_result=False,
cache_expiration=timedelta(
minutes=2,
),
)
def fetch_temp(lat: float, lon: float):
base_url = "
https://api.open-meteo.com/v1/forecast/"
weather = httpx.get(
base_url,
params=dict(
latitude=lat,
longitude=lon,
hourly="temperature_2m",
),
)
most_recent_temp = float(weather.json()["hourly"]["temperature_2m"][0])
return most_recent_temp
@task(
retries=3,
cache_expiration=timedelta(
minutes=10,
),
)
def fetch_wind(lat: float, lon: float):
base_url = "
https://api.open-meteo.com/v1/forecast/"
weather = httpx.get(
base_url,
params=dict(
latitude=lat,
longitude=lon,
hourly="windspeed_10m",
),
)
most_recent_wind = float(weather.json()["hourly"]["windspeed_10m"][0])
return most_recent_wind
@task
def save_weather(temp: float):
with open("weather.csv", "w+") as w:
w.write(str(temp))
return "Successfully wrote temp"
@flow
def pipeline(lat: float, lon: float):
logger = logging.get_run_logger()
temp = fetch_temp(lat, lon)
result = save_weather(temp)
x = fetch_wind(lat, lon)
logger.info(f"temp >>>>>>>>> {temp }")
logger.info(f"wind >>>>>>>>>> {x}")
result = save_weather(x)
return result
if
name == "__main__":
pipeline(38.9, -77.0)