https://prefect.io logo
a

Abhilash Agarwal

08/16/2023, 4:45 PM
import httpx
from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta
@task(retries=2)
def fetch_weather(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])
print(f"Most recent temp C: {most_recent_temp} degrees")
return most_recent_temp
@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(minutes=5))
def save_weather(temp: float):
with open("weather.csv", "w+") as w:
w.write(str(temp))
return "Succefully wrote temp"
@task(retries=2, retry_delay_seconds=2)
def windspeed(lat: float, lon: float):
base_url = "<https://api.open-meteo.com/v1/forecast/>"
wind = httpx.get(
base_url,
params = dict(latitude=lat, longitude=lon, hourly="temperature_2m"),
)
wind_speed = float(wind.json()["hourly"]["windspeed_10m"][0])
print(f"Current wind speed C: {wind_speed}")
return wind_speed
@flow(retries=3)
def pipeline(lat: float, lon: float):
temp = fetch_weather(lat, lon)
result = save_weather(temp)
curr_wind = windspeed(lat, lon)
print(f"Current temp C: {temp} degrees and windspeed is {curr_wind}")
if __name__ == "__main__":
fetch_weather(38.9, -77.0)
(venv) abhilashagarwal@Abhilashs-MacBook-Air venv % python weatherdata.py Traceback (most recent call last): File "/Users/abhilashagarwal/Desktop/Prefect_Training/venv/weatherdata.py", line 43, in <module> fetch_weather(38.9, -77.0) File "/Users/abhilashagarwal/Desktop/Prefect_Training/venv/lib/python3.11/site-packages/prefect/tasks.py", line 533, in call return enter_task_run_engine( ^^^^^^^^^^^^^^^^^^^^^^ File "/Users/abhilashagarwal/Desktop/Prefect_Training/venv/lib/python3.11/site-packages/prefect/engine.py", line 1117, in enter_task_run_engine raise RuntimeError( RuntimeError: Tasks cannot be run outside of a flow. To call the underlying task function outside of a flow use
task.fn()
.
Seeing this error when I try to run my code. Any suggestions?
m

Mikalai

08/16/2023, 4:49 PM
Copy code
if __name__ == "__main__":
    pipeline(38.9, -77.0)
a

Abhilash Agarwal

08/16/2023, 4:49 PM
Yup I just saw it too, thank you so much!