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

Prefect Practitioner Cert May 16 2023 - 101 .pdf

i'm getting an error: PDF document is damaged

<@U03KWSKFN8L> we're stuck not knowing what to do because we can't open the pdf :slightly_smiling_face:

<@U03KWSKFN8L> Can you share the github link

<https://github.com/discdiver/ppcc-2023-05>

`<https://api.open-meteo.com/v1/forecast?latitude=52.52&amp;longitude=13.41&amp;daily=temperature_2m_max,temperature_2m_min>`

101- Prefect Practitioner Cert May 16 2023.pdf

Yes. I don’t see why. I shared the Google slides below.

Sorry these pdfs are corrupted for some reason. working on it.

Prefect Practitioner Cert May 16 2023.png

set the channel topic: Tech Setup: <https://docs.google.com/document/d/136OtdLNrYOiYsOXTYPluxnFlnkyWpyZ08rp-7sq6PFc/edit?usp=sharing>

Zoom:
<https://us06web.zoom.us/j/89594449306?pwd=bm9TdWs4ZGRGS083bGhTTlYvZTRGdz09>

Passcode 355535

We will get back together at 11:10am ET. Feel free to take a break if you are done.

Should we leave the breakout room when done?

Yep! Once you’re done head back to the main room, and we will start here shortly - grab some more coffee if you need!

Celery is a task queue, while Prefect is an orchestration and observation library and platform. Prefect’s scope is quite a bit longer. We’ve heard that some folks use Prefect as a Celery replacement.

<https://docs.prefect.io/latest/concepts/tasks/#task-arguments>

wrt retries and exponential backoff, the docs have it covered: <https://docs.prefect.io/latest/concepts/tasks/#retries>

<https://docs.google.com/presentation/d/1VLKxXnIMKRyUvJJ8muwnE5xKBR-5ZW7IkNzLGr7bBfc/edit?usp=sharing|101 Slides> in Google Sides

Prefect Practitioner Cert May 16 2023 - 102.pdf

Logging docs: <https://docs.prefect.io/latest/concepts/logs/#logging>

If you need to clear your server database: `prefect server database reset`

Result serializer: <https://docs.prefect.io/latest/concepts/results/?h=results#result-serializer>

Task Runners: <https://docs.prefect.io/latest/concepts/task-runners>

image.png

Hello, I am trying to use prefect server on an a cloud IDE (GitPod) that assigns an hostname. Clearly 127.0.0.1 makes no sense as an API Endpoint

<https://github.com/gevent/gevent/issues/1721#issuecomment-1223751954|M1 Greenlet issue>

<https://github.com/PrefectHQ/prefect/issues/7978> <@U03DK3VVAMD>

```import httpx
from prefect import flow, task


@task(retries=3, retry_delay_seconds=5)
def fetch_temperature(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, retry_delay_seconds=5)
def fetch_humidity(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="relativehumidity_2m"),
    )
    most_recent_hum = float(weather.json()["hourly"]["relativehumidity_2m"][0])
    return most_recent_hum


@flow
def convert_c_to_f(temp: float):
    return temp * 9 / 5 + 32


@task
def save_weather_data(temp: float, variable_name: str):
    with open(f"weather_{variable_name}.csv", "w+") as w:
        w.write(str(temp))
    return f"Successfully wrote {variable_name}"


@flow
def pipeline(lat: float, lon: float):
    temp_c = fetch_temperature(lat, lon)
    temp_f = convert_c_to_f(temp_c)
    print(temp_f)
    temp_result = save_weather_data(temp_f, "temperature")

    humidity = fetch_humidity(lat, lon)
    print(humidity)
    rain_result = save_weather_data(humidity, "humidity")

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

Room 2

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


@task(retries=3, retry_delay_seconds=10)
def geocode(name: str):
    base_url = "<https://geocoding-api.open-meteo.com/v1/search>"
    geocoded = httpx.get(
        base_url,
        params=dict(name=name),
    )
    first_result = geocoded.json()["results"][0]
    return (first_result["latitude"], first_result["longitude"])


@task(retries=3, retry_delay_seconds=10)
def fetch_weather(lat, lon):
    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, retry_delay_seconds=10)
def fetch_historical_weather(start, end, lat, lon):
    base_url = "<https://archive-api.open-meteo.com/v1/archive>"
    weather = httpx.get(
        base_url,
        params=dict(latitude=lat, longitude=lon, start_date=start, end_date=end, hourly="temperature_2m"),
    )
    breakpoint()
    most_recent_temp = weather.json()["hourly"]["temperature_2m"]
    return most_recent_temp

@task(retries=3, retry_delay_seconds=10)
def fetch_elevation(lat, lon):
    base_url = "<https://api.open-meteo.com/v1/elevation>"
    elevation = httpx.get(
        base_url,
        params=dict(latitude=lat, longitude=lon),
    )
    elev = elevation.json()["elevation"]
    return elev

@flow
def current_weather(name: str):
    logger = get_run_logger()
    geoname = geocode(name=name)
    temperature = fetch_weather(*geoname)
    elevation = fetch_elevation(*geoname)
    <http://logger.info|logger.info>(f"temperature={temperature}")
    <http://logger.info|logger.info>(f"elevation={elevation}")
    return temperature, elevation

@flow
def historical_weather(name: str, start_date: str, end_date: str):
    logger = get_run_logger()
    geoname = geocode(name=name)
    lat, lon = geoname
    temperature = fetch_historical_weather(start=start_date, end=end_date, lat=lat, lon=lon)
    <http://logger.info|logger.info>(f"historical temperature={temperature}")
    return temperature


if __name__ == "__main__":
    current_weather("Rome")
    historical_weather("Rome", "2022-11-31", "2022-12-31")```

<https://docs.prefect.io/latest/cloud/users/roles/>

<https://docs.prefect.io/latest/concepts/blocks/>

Can you give a use-case example of Blocks vs Variables?

They are very similar in general; an example that I have used, are static API urls for a 3rd party api.
There are routine messages sent in a flow to this 3rd party api (jira) , but I have different api URLS between tiers (dev / stage / prod) . I use the same code in the flow to reference the same named variable, but different values across workspaces

e.g., dev might be: `https:/dev.jira.com` then `<https://stage.jira.com>` where the flow code references just a single name variable

then if it changes, I can just change it in a single location

Blocks are preferred over Variables for storing sensitive information, right?

yes, specifically SecretStr which are intentionally obfuscated

<https://docs.prefect.io/latest/concepts/blocks/#secret-fields>

An alternate mechanism is using a JSON block to store a reference / location to a secret in your existing secrets manager like AWS / Azure Keyvault

you can defer the secrets that way, but that requires a little more hand-written code to access

<@U03DK3VVAMD> - I have some examples using ACI and AKS that use a private image registry, and private github repository, that do not use credentials stored in Prefect, if that helps address your question

These use ImagePullSecrets, and Managed Identity to access