https://prefect.io logo
Title
w

Wesley Jin

03/01/2022, 11:45 PM
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
a

Anna Geller

03/01/2022, 11:53 PM
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.
🙏 1
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"])])
w

Wesley Jin

03/02/2022, 12:01 AM
thanks @Anna Geller for the context! good to know about the other ways 🙂
a

Anna Geller

03/02/2022, 12:01 AM
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:
    ...