What's the recommended way to set labels for a flow?

[anna_geller]

View in #prefect-community on Slack

Question

@Wesley_Jin: What difference, if any, is there between the labels set during flow.register vs. on the run_config object e.g. ECSRun ? If not, is there a preferred place to set it?
curious - as our team building out a CI pattern where each flow in our repo gets a default storage & run_config (if not set already) & registered to Prefect Cloud. We have ECS agents running in each of our environments (w/ the appropriate labels) to pick up on each environment’s flows

Answer

@Anna_Geller: It’s a really good question and I totally get why this may be confusing

Prefect is generally un-opinionated about it and provides you several ways to configure this so that you can incorporate it into your use case as you see fit.

I personally think that setting labels on run config is the cleanest, but there is no right or wrong here.
To confuse you even further those are all ways how you can set labels that I’m aware of:

1. Run config
2. Register CLI or flow.register
3. in the UI (see image below)
4. on a Clock (e.g. on a CronClock) - see code below

from prefect.schedules import Schedule
from prefect.schedules.clocks import CronClock

schedule = Schedule(clocks=[CronClock("0 0 * * *", labels=["some-label"])])

@Wesley_Jin: thanks @Anna_Geller for the context! good to know about the other ways

@Anna_Geller: and since you asked how you can set default run config and default storage in a way to avoid code duplication - check out this blog post. Here is example from it showing this:

from prefect.run_configs import LocalRun, KubernetesRun, RunConfig
from prefect.storage.github import GitHub
from prefect.client.secrets import Secret


def set_run_config(local: bool = False) -> RunConfig:
    if local:
        return LocalRun(labels=["dev"])
    aws_account_id = Secret("AWS_ACCOUNT_ID").get()
    return KubernetesRun(
        labels=["prod"],
        image=f"{aws_account_id}.<http://dkr.ecr.us-east-1.amazonaws.com/prefect-dbt-k8s-snowflake:latest|dkr.ecr.us-east-1.amazonaws.com/prefect-dbt-k8s-snowflake:latest>",
        image_pull_policy="IfNotPresent",
    )


def set_storage(flow_name: str) -> GitHub:
    return GitHub(
        repo="anna-geller/prefect-dbt-k8s-snowflake",
        path=f"flows/{flow_name}.py",
        access_token_secret="GITHUB_ACCESS_TOKEN",
    )

Usage in a Flow:

from prefect import Flow
from flow_utilities.prefect_configs import set_run_config, set_storage


FLOW_NAME = "02_dbt_snowflake"

with Flow(FLOW_NAME,
          storage=set_storage(FLOW_NAME),
          run_config=set_run_config(),
          ) as flow:
    ...

https://www.prefect.io/blog/orchestrating-elt-on-kubernetes-with-prefect-dbt-and-snowflake-part-2/