Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Traceback (most recent call last):
  File "/opt/homebrew/Caskroom/miniconda/base/bin/prefect", line 5, in &lt;module&gt;
    from prefect.cli import app
  File "/opt/homebrew/Caskroom/miniconda/base/lib/python3.11/site-packages/prefect/__init__.py", line 45, in &lt;module&gt;
    from prefect.engine import pause_flow_run, resume_flow_run
  File "/opt/homebrew/Caskroom/miniconda/base/lib/python3.11/site-packages/prefect/engine.py", line 98, in &lt;module&gt;
    from anyio import start_blocking_portal
ImportError: cannot import name 'start_blocking_portal' from 'anyio' (/opt/homebrew/Caskroom/miniconda/base/lib/python3.11/site-packages/anyio/__init__.py)

`docker run -it --rm -v $PWD:/opt/prefect/flows/ prefecthq/prefect:2-latest bash`

`import httpx`
from prefect import flow, task
import pandas as pd
from prefect.tasks import task_input_hash

# latitude 26.1 North, longitude 84.8 West

# TASKS

# cache_key determines the inputs
@task(cache_key_fn=task_input_hash, cache_expiration=datetime.timedeltas=)
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,cloudcover_mid,windspeed_10m,winddirection_10m",
        ),
    )
    weather_json = weather.json()
    df = pd.DataFrame(
        {
            "latitude": weather_json["latitude"],
            "longitude": weather_json["longitude"],
            "temperature": weather.json()["hourly"]["temperature_2m"],
            "wind": weather.json()["hourly"]["windspeed_10m"],
            "cloudcover": weather.json()["hourly"]["cloudcover_mid"],
        }
    )
    return df


@task
def save_weather_data(data: pd.DataFrame):
    <http://data.to|data.to>_csv("weather_data.csv")
    return "successfully wrote file"


@task
def get_latest_value(data: pd.DataFrame, metric: str):
    print(f"latest data for {metric} {data[metric].iloc[-1]}")

# FLOW
# Determines the order in which tasks are called

@flow(retries=4)
def get_weather_metrics(lat: float, lon: float):
    df = fetch_weather(lat=lat, lon=lon)
    # print out latest wind metric
    get_latest_value(df, "wind")
    save_weather_data(df)

# Start the flow
if __name__ == "__main__":
    get_weather_metrics(38.9, -77.0)

Issue was a syntax error:
```@task(cache_key_fn=task_input_hash, cache_expiration=datetime.timedeltas=)```

Thanks for helping with the discovery of another bug/fixing team :slightly_smiling_face:
Btw, have you come across this issue with deployment? RuntimeError: Docker worker cannot be used with an ephemeral server.

Hmm… no I haven’t seen that one. Let me take a look…

Hmm, ok I <https://github.com/PrefectHQ/prefect-docker/blob/797bf288c604308a4b68493fe573a84b42b1eebb/prefect_docker/worker.py#L382-L385|see>. So you need to either be pointing to cloud or have a local server running.

So in another terminal run `prefect server start`

Then in your original terminal run `prefect config set PREFECT_API_URL=<http://127.0.0.1:4200/api>`

Then you should be able to start the worker.

Or you can log into cloud and that will have the same effect