is anyone else seeing the following mypy error?

Thread for Tom, Akhil, Kirthan, Tim and Kalo

Thread for Jon, Marius, Quan, Emiliano

Sample basic code to fetch

windspeed_10m

import httpx  # requests capability, but can work with async
from prefect import flow, task


@task
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])
    return most_recent_temp

@task
def fetch_windspeed(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_windspeed = float(weather.json()["hourly"]["windspeed_10m"][0])
    return most_recent_windspeed

@task
def save_weather(temp: float):
    with open("weather.csv", "w+") as w:
        w.write(str(temp))
    return "Successfully wrote temp"


@task
def save_windspeed(temp: float):
    with open("windspeed.csv", "w+") as w:
        w.write(str(temp))
    return "Successfully wrote temp"




@flow
def pipeline(lat: float, lon: float):
    temp = fetch_weather(lat, lon)
    result = save_weather(temp)
    return result


if __name__ == "__main__":
    pipeline(38.9, -77.0)

Thread from breakout 1

playing around for exercise 1:

from enum import Enum, auto

import httpx  # requests capability, but can work with async
from prefect import flow, task


class TimeResolution(Enum):
    hourly = auto()
    daily = auto()


@task
def fetch_weather_hourly_param(lat: float, lon: float, resolution: TimeResolution, param: str) -> dict:
    base_url = "<https://api.open-meteo.com/v1/forecast/>"
    weather = httpx.get(
        base_url,
        params={
            "latitude": lat,
            "longitude": lon,
            resolution.name: param
         }
    )
    return weather.json()


@task
def save_weather(temp: float):
    with open("weather.csv", "w+") as w:
        w.write(str(temp))
    return "Successfully wrote temp"


@task
def extract_data_point(response: dict, resolution: TimeResolution, param: str, hour: int) -> float:
    return float(response[resolution.name][param][hour])

@task
def print_float_param(data: float, resolution: TimeResolution, param: str):
    print(f"for param {param}, in resoltion {resolution.name} data is {data}")


@flow
def pipeline(lat: float, lon: float, resolution: TimeResolution, item_pos: int, params: list):
    params_jsons = fetch_weather_hourly_param.map(lat, lon, resolution, params)
    float_data = extract_data_point.map(params_jsons, resolution, params, item_pos)
    return print_float_param.map(float_data, resolution, params)


if __name__ == "__main__":
    params = ["temperature_2m", "relativehumidity_2m", "dewpoint_2m"]

    pipeline(38.9, -77.0, TimeResolution.hourly, 0, params)

For more on subflows and design patterns in general: • https://www.prefect.io/guide/blog/dataflow-design-patterns/ • https://github.com/taylor-curran/flow-patterns/tree/main/flows/subflows

Lab 102 Instructions • Make a new flow that grabs different weather data from open-meteo • Add retries • Add a subflow that transforms or aggregates the data

If you need to change your permissions to share your screen and you get booted from Zoom, that’s fine. Just let me know and I’ll let you back into your breakout room.

Share your code for module 2 here: (thread)

how do you approach the decision of whether to write a new task or a new subflow?

Let’s get back together at 10:45am.

I joined a bit late. Did I miss creating anything from the beginning?

I am getting a weird error message when I try to run prefect server locally:

10:37:45.809 | ERROR   | prefect.server.services.telemetry - Unexpected error in: OperationalError('(sqlite3.OperationalError) database is locked')

fyi, thought this was fun:

def distance_to_collector(lat,lon):

params = {"latitude":lat, "longitude":lon, "hourly":"temperature_2m"}

result = httpx.get("<https://api.open-meteo.com/v1/forecast/>",params = params).json()

collector_lat = result['latitude']

collector_lon = result['longitude']

return haversine.haversine((lat,lon),(collector_lat,collector_lon),unit=haversine.Unit.FEET)

Gravatar: https://en.gravatar.com/

Prefect APIs info: https://docs.prefect.io/2.10.11/api-ref/

103 Lab Instructions • Use a previously made flow • Use Prefect Cloud • Make an email notification automation for flow run completion • Run the flow a few times • Check out the event feed in the UI

image.png

is it possible to move flows from one workspace to another?

Emil shared useful commands: see if you're using prefect cloud

prefect version

see active workspace

prefect cloud workspaces ls

What happens if one of the automation or blocks is failed while running a flow? Is this logged somewhere? can be reflected in the flow results somehow?

I'm trying to have a better understand of Secrets block. Say I have a k8s cluster at one of my customers, Inside there the k8s cluster there is a hashicorp vault service (not cloud). Can i set up a secret who's secret- provider is this vault k8s service and non of those secrets are actually kept in prefect cloud ?

What time are we gathering again?

Hello! I was wondering what the Retrying state is in automations?

is it possible to update (e.g. add a new key-value pair) to a JSON block, or is that not really the intended use-case?

can you put tomorrow slides in advance here?

How do you unregister a block? What would be the opposite of https://docs.prefect.io/2.10.11/concepts/blocks/#registering-blocks-for-use-in-the-prefect-ui?