Has anybody consume Prefect api in C# application?

How can I setup in Orion the task runner at deployment? I want to be able to change the mode between local dev and prod. I can only see how to do that for flow runners at the deployment level

hi all, maybe a newbie Orion question - how can I use more than one Python file to describe a flow with multiple tasks (we've got tons of internal ML libraries)? apart from putting all those in a Docker image, I haven't found a way for now

Can we schedule each task at a different time inside a single flow? Example: a Flow is cronned for 3pm and it has TASK A, B, C. Task A should start at 3 pm but Task B and Task C should start at 4:30 pm and 5pm respectively.

Hello, I have some questions. Please look at my figures. I understood Prefect works like figure 1(1 Pod or container for Agent) with autoscailing by Dask. Did I understand correctly? Plus, is there a way to use Prefect like figure 2(N Pods for flow)

Hello. Is it possible using prefect.Client or graphql get upstream task_run_ids by task_run_id?

Hey all! Quick reminder that PrefectLive is happening today at 3p Eastern with Chris White. Hope to see you on Twitch! https://prefect-community.slack.com/archives/C036FRC4KMW/p1647289203959699

Hey All I have a flow with "*state_handlers*" to slack and some api. Regarding that I have 2 questions: 1. When I change the flow's state from the Ui to Failed or Cancelled , I don't get any message from the handler , it's like it doesn't trigger the state handlers, Am I correct? 2. I want to create API call to the prefect server to change state or cancel flow_run, is it supposed to trigger the state handlers?

Hello everyone! I have a question about Orion s3 storage. When I created a test deployment, a file was created on my bucket, but when I tried to run the deployment, I got this error(in the thread) Of note: I use corporate AWS account and do not have full S3 Access, could this be the core of the issue?

we just (yesterday?) started running into a cloudpickle error when starting flow runs:

prefect.engine.signals.FAIL: Flow finished in state <Failed: "Failed to load and execute flow run: FlowStorageError('An error occurred while unpickling the flow:\n  AttributeError("\'Comment\' object has no attribute \'_end\'")')">

the flows are stored in GCS, python is 3.9.10 both locally (MacOS) and remote (docker+k8s). same prefect+cloudpickle versions. I've had the same thing happen before when python versions don't match but bumping has always fixed it, not really sure what else to look into now. any suggestions...?

Can I reuse a single Docker storage object for multiple registered flows?

Hi Prefect Experts - Is it possible to not display generated tasks like

List

and

Dict

from the visual schematic displayed in the UI?

Highlighting this morning's news here - today we announced Club 42, our ambassador program! Some of our most experienced and dedicated community members have come together to help the community grow and thrive. 🥂 Here's to the P community! https://prefect-community.slack.com/archives/CKNSX5WG3/p1647439185450549

Hi, I'm trying to run DataBricks Tasks using a docker agent. However the flow is failing when importting de DB tasks. When writting from prefect.tasks.databricks.databricks_get_job_id import DatabricksGetJobID, it fails with: 'ModuleNotFoundError: No module named 'pydantic''. I'm using the latest-python3.8 docker image

Hi, I have a flows.json that contains the metadata for a flow. I want to register this flow inside of another prefect flow. Is that possible?

hello, i am having issues with installing prefect 2.0 in my aws. I have a ubuntu 18.0 running which comes with a older version of sqlite3. Prefect 2.0 uses version greater than 3.24. is there anyway i can not install the sqlite3 depdency when installing prefect? @Kevin Kho @Anna Geller

Hi all! 👋 Sorry for the late cancellation but we had to move PrefectLive to tomorrow at 2p Eastern as @Chris White had some issues come up. Apologies again, I know you probably had your popcorn popped but we'll see you LIVE 📺 in under 24 hours.

Hi, I am running on prefect UI and getting this attribute error: can't get attribute 'get_target_name" on <module prefect_scripts.tasks.trade_cache_tasks'. the trade_cache_task is like below, similar structure works fine on other scripts, anyone has any ideas?

def get_target_name(x, **kwargs):
	…
	return target_name

def get_task_name(x, **kwargs):
	…
	return task_name

@task(task_run_name=get_task_name,target=get_target_name,checkpoint=True,result=LocalResult())
def task_a(x, y, z):
	…
	return …

Want to check if what I read about orion is accurate... Right now my team runs a bunch of tasks via cron jobs. Would it be possible to start with 2.0 by only using the scheduler aspect? It'd be cool to have a UI for what has run. edit: I don't have experience with 1.0 or 2.0

Hello, signed up for cloud beta but now stuck on the new workspace page. When I am trying to create a new workspace, everything is greyed out. Do I need to complete any additional steps to make this work ?

I've built a flow_of_flows orchestration for my prefect project. Is it possible to see the task dependencies under each nested flow in the UI schematic? Right now I just see 2 flows, but I can't seem to drill down further into each flow

Hi! For Kubernetes agent, is there a way to set environment variables in the Kubernetes jobs that are spun up by prefect agent pod?

So I have an ETL flow that I'm executing (shown below). There is a task named

execute_ddls

that only executes if

new_ddl_exist

is True. When

execute_ddls

is skipped, all my downstream dbt tasks are skipped but regardless if the case statement is entered, I want all downstream tasks to run. I tried

skip_on_upstream_skip = False

on the dbtShellTask but it doesn't work. I'm sure I'm missing something so trivial...

with Flow('bi_test_flow') as flow:

    # new DDLs (if any)
    ddls = new_ddls_to_run(loaded_files, os.listdir(DDL_PATH))

    # # # execute new DDLs ONLY if they exist
    new_ddl_exist = do_new_ddl_scripts_exist(ddls)
    with case(new_ddl_exist, True):
        execute_ddls = execute_sql(ddls)

    dbt = dbt(
        command="dbt run -m anlyz_base.views",
        upstream_tasks=[execute_ddls],
    )
    dbt_operations = dbt(
        command="dbt run-operation materialize_views"
    )

Hi, I have added state_handlers to my flow, but for some reason they are not getting triggered. Currently it looks like this:

from prefect.utilities.notifications import slack_notifier
.
.
flow.state_handlers = [slack_notifier()]

I have added the prefect app and also added the webhook URL to the

SLACK_WEBHOOK_URL

secret. Any pointers on why this is happening / how to debug?

Hi again, I have a question about extra logging. We have own Python module 'bear' whereas logging is defined as:

logger = logging.getLogger(__ _name ___)

and I want log messages from that module to appear in Prefect. I registered extra logger, and log messages defined in Task (AWSPOC) appear, but not the log messages defined inside 'bear'. My code for "prefect_flow.py" :

import logging

format_string = "%(asctime)s %(name)s [%(levelname)s] %(message)s"
logger = logging.getLogger('magic_logger')
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
handler.setLevel(logging.DEBUG)
formatter = logging.Formatter(format_string)
handler.setFormatter(formatter)
logger.addHandler(handler)

class AWSPOC(Task):

    def __init__(self, name: str, config_file: str):
        self.config_file = config_file
        super().__init__(name=name)

    def run(self):
        configuration = load_file(self.config_file)
        loader = Loader(configuration=configuration)
        <http://logger.info|logger.info>("This message appears in Prefect log output")
        loader.run()

executor = LocalDaskExecutor()
host_config = {...}
storage = S3(...)
env = {
    "PREFECT__LOGGING__EXTRA_LOGGERS": "['magic_logger']"
}
docker_run_config = DockerRun(image=..., host_config=host_config, env=env)

with Flow(name="AWS POC", executor=executor, storage=storage, run_config=docker_run_config) as flow:
    task = AWSPOC(name='...', config_file='...')
    task()

flow.register(project_name='AWS POC')

Where I'm wrong? The only idea I have is that 'bear' should have it's own named logger.

Hi. 1. Is it possible to specify timeout for wait_for_flow_run task? 2. Is it possible to use prefect.Client to cancel Prefect flows by name?

I would like to connect my prefect service, which is deployed to gcp kubernetes engine using the helm chart, with my postgres database hosted in gcp cloudSQL. Is there any documentation or tutorial for doing this?

Hello, I am wondering if it is possible to set up custom results classes with custom serialisers in Prefect 2.0?

I have an interesting situation. I'm pulling data from an API. I can pull only 100 items at a time (there are like 200k total). The API response gives me a "next url" that I can use in the next API request. However, that "next url" is the same as I iterate through (I assume it's some sort of stream). Once I get the results, I want to write them to a database. My non-prefect way is to write a task that pulls from the API until there is no next_url (

while has_next_url: ...

) . Is there a better "prefect" way to do this? One of the downsides of my basic way is that, if there is an error along the way (let's say at record 199,999 of 200,000), I lose what I had.

Sorry if this has been asked/answered before. One thing I really miss from other orchestrators is a simple way to specify how concurrency for a flow should behave. In Airflow I believe this is controlled with using a

LatestOnlyOperator

or using

depends_on_past

. In Oozie (I know, not my favorite orchestrator), this could be achieved by setting a

concurrency

to

1

and

execution

to

FIFO

or

ONLYLAST

. Are there any plans in Orion to provide similar functionality so users don’t have to implement custom solutions to this common pattern?