Is it possible to sign in via password auth instead of the magic link? I am no longer seeing that option

Hi, We have lot of stuff in Prefect 1.0, We are not planning to migrate to Prefect 2.0 anytime soon. is there any plan for deprecation of Prefect 1.0, will there be any impact if we are not migrating to 2.0?

Hey folks, I believe I’ve found a bug with the

allow_failure()

wrapper for the

KubernetesJob

infra - it seems to have no effect when running with

SequentialTaskRunner

or

ConcurrentTaskRunner

. When I run the same flow locally it works as expected (downstream tasks are ran), but when it’s ran on k8s (with

KubernetesJob

infrastructure) it ignores the wrapper and the downstream tasks just go to

NotReady

as if the

allow_failure()

wrapper weren’t used at all. MRE in 🧵

We're running an orion instance. Since orion is not using an externally hosted API,

PREFECT_API_URL

is not set (

PREFECT_ORION_UI_API_URL

is).

printenv | grep -E "PREFECT.*API"
# PREFECT_ORION_UI_API_URL=<https://domain.tld/api>

FlowRunNotifications use

PREFECT_API_URL

as base api url and since

PREFECT_API_URL

is not set

api_url

defaults to a rather useless

<http://ephemeral-orion/api>

. I'm wondering if it would make sense to change this line to use

PREFECT_ORION_UI_API_URL

if

PREFECT_API_URL

is not set. I see that

PREFECT_ORION_UI_API_URL

defaults to

PREFECT_API_URL

, but I think if I set

PREFECT_API_URL

to a seemingly external url api calls from ui will stop being routed through loopback and that would be a whole thing.

Hi friends, I would like to make a common

create_deployment.py

utility that my CI/CD uses that will create and apply deployments when flows change. Instead of complicated bash, it’s easier to do it in Python. However, I’m unclear how I can get the

flow

object from a filepath in this case:

deployment = Deployment.build_from_flow(
    flow=my_flow, # how do i get from a filepath to the flow
    name="s3-example"
)

In Prefect v1, there was a useful utility called

from prefect.utilities.storage import extract_flow_from_file

— my question is does such a utility exist in v2? Or alternatively, is there a better way of doing this?

Hello! is there a way to kill a flow run? I have a flow running as a kubernetes job and it fails because the image does not exist. I deleted the flow run from the UI and killed the job in k8s but a new job keeps being created. Thanks!

👋 hullo! since 2.3.0 it appears we can add flow code directly into your docker image, and prefect won't need a remote storage block for a

DockerContainer

. I'm wondering, does that same logic apply to a

KubernetesJob

(since we also have to provide an image for that infra block)?

Is anyone working on a Google Cloud Batch backend? If not and I were interested in working on one, where would you point me as a starting place?

Hello! I'm working on setting up a demo for prefect to programmatically build a deployment and run it in a docker container. I have a single script (see script in thread) that saves a storage block, an infrastructure block, builds a deployment, applies it, and attempts to run it with

run_deployment

. When running the script, I get

JSONDecodeError: Expecting value: line 1 column 1 (char 0)

that comes from

run_deployment

and is thrown at

prefect/client/orion.py", line 1329, in read_deployment_by_name: return schemas.core.Deployment.parse_obj(response.json())

. I edited

orion.py

to add

print(response.json())

at line 1329 and then that line threw the error, but the json response still prints and it looks like a valid json response object. I'm using prefect 2.6.4. I would appreciate any help in getting the deployment working, thanks!

I have prefect agent running on Lambda that connected to a VPC, at the moment I allow all traffic from everywhere for the inbound Security group setting, but this is not ideal, besides port 4200, might I know what other ports I need to allow to get Prefect running on Lambda with a VPC without issue? Thank you.

Quick Q regarding task dependencies in Prefect 2.x: In Prefect 1.x I could pass `upstream_tasks=[task_a, task_b]`` to a task in a flow. What is the equivalent in Prefect 2.x?

Hi Team, Is it possible to run an agent for two different workspaces for one container? Can we have multiple profiles inside a container? I don't think it is possible, because the Prefect CLI will be pointing to a specific workspace, but I thought I'd ask instead of just spinning up two ECS instances.

Can I use both .map and .submit on a task?

Hello again! is there a way to not display flow run parameters of type pydantic.SecretStr in the UI?

Hello Team, is there a reason parameter entered in UI doesn't work in the run? When I make a change to the parameter in the UI, the flow still picks up the default parameter instead of the newly entered one.

hey all, im trying to pass a task's result to a function that's not a task. I'm getting:

TypeError: not all arguments converted during string formatting

is there a way to get the parameter's value without giving function with a task decorator?

"attempted to send logs to Orion without a flow run id": We are running into an issue where our flow imports libraries that do some logging on import. As outlined here: https://linen.prefect.io/t/3323130/must-get-run-logger-be-called-in-each-task-flow-i-have-a-flo Is there a known work around? We tried to disable logging & re-enable logging post imports, but that didn't help.

Hey, congrats on releasing 2.6.5. Why is it, that for manual retries via UI we need to enable result persistence? Don't we simply rerun a flow - why do we need the results from the failed flow to be persisted? (Not that it's a problem, I'd just like to understand the underlying architecture)

Hello all, After my last problem was immediately and perfectly solved here in the channel (thank you, again, @Q !), I would like to ask if anyone could help me with a second question, concerning the same example project (https://prefect-community.slack.com/archives/CL09KU1K7/p1666555308227089)? Namely, now that the creation of the deployment on the SFTP remote storage block works, I'm unfortunately unable to start a flow run from it. Specifically, I started a local Prefect agent and afterwards a local flow run from the UI (Run -> Now with defaults). After that, the paramiko library gives the following error:

File "C:\Users\Testbenutzer\anaconda3\envs\prefect_test_project_v3\lib\site-packages\paramiko\sftp_client.py", line 810, in get
    with open(localpath, "wb") as fl:
PermissionError: [Errno 13] Permission denied: '.'
23:16:29.384 | INFO    | prefect.infrastructure.process - Process 'upbeat-crab' exited cleanly.

To solve the PermissionError, I tried two approaches, which unfortunately both failed: 1. Try to change the

PREFECT_LOCAL_STORAGE_PATH

2. Running Powershell, from which I start Prefect, as admin My problem seems to be practically the same as that reported by @Mike He in the discourse forum (https://discourse.prefect.io/t/agent-debugging-agent-execution-error-with-ftp-storage-in-windows/1587) and here in Slack (https://prefect-community.slack.com/archives/C03D12VV4NN/p1663319368100279). (The only difference I see is that in my case, the last error of the traceback comes from the paramiko library instead of fsspec.) I saw that he already developed a workaround, which however includes changing the Prefect source code which I have not tried yet. I would write details, e.g. the full traceback, in the thread. If anyone had an idea or a hint how this problem could be solved, preferably without manually changing the Prefect source code, you would help me again a lot. Thank you very much in advance!

Hi. Does anybody had any issues with Prefect 1.x ECS deployment ? Regarding the flow register not having the Fargate machine you requested ? I registered 16vCPU and have in the task definition only a 2vCPU machine. Spoke with AWS support but it was not helpful and our AWS Account does have a vCPU limit at 2000. So enough bandwidth.

Hi - I have my task and flow in seperate files, but would like to use this

def cache_within_flow_run(context, parameters):
    return f"{context.flow_run_id}-{task_input_hash(context, parameters)}"

How would i pass the flow_run_id into the task this function and further into the task decorator?

@task(cache_key_fn=cache_within_flow_run)
def cached_task():
    print('running an expensive operation')
    return 42

Hi, is it possible to add the ECR as a block in Prefect 2? I was wondering if this is doable, I'm was trying in the UI, but the credentials part got me thinking. I'm using IAM Roles to authenticate the flow runs, so I don't even have an ECR username and password which the Docker Registry block in the UI requires. Maybe someone knows a way to achieve that. Thanks in any case

i had a conceptual question about the deployment running infra. when we run same deployment multiple times (with same or different params) does each of them create new pods to run new job or are these pods created specific to flows ? and if i want my parallel processing(parallel running tasks or flows) to be distributed into multiple jobs so as to make sure the amount of resources each process is taking is not huge.. how can i achieve that ? is creating separate deployments a way to do it ?

We are on the latest prefect & prefect_aws, and seeing scheduled jobs terminate without being run (intermittently). The UI for the flow run shows no logs, no task runs, and no subflow runs. In cloudwatch logs for the flow run, I found the below. Any thoughts?

23:02:20.896 | INFO    | prefect.engine - Engine execution of flow run '682567a2-e507-4aa1-b5c7-25bb2cc2573f' aborted by orchestrator: This run has already terminated.

quo https://docs.prefect.io/concepts/deployments/#parameters-in-deployments it's possible to add default parameters in yaml and python deployments... Is it possible or in the works, to also add these via the cli deploy?

hey there, i'm trying to use a task's result as a directory name. is this possible?

Hello, I have a flow that I run on my laptop using DockerContainer infrastructure and takes around 4 minutes to complete. When I run that same flow on a kubernetes cluster using KubernetesJob it times out after 30 minutes (limit that I set in my flow). Any tips on how I could troubleshoot what is going on? I am using the default job template.

Hi, I'm running a task that creates a flow run from a deployment ID using create_flow_run_from_deployment. I used to run the following code in prefect 2.5.0:

await my_task.submit(_flow_name_=FLOW_NAME, _env_=env, _run_name_=run_name, _wait_for_=[preprocess], _parameters_=parameters)

However after upgrading to 2.6.5 my_task isn't even being run. My workaround is to just run this:

await my_task(_flow_name_=FLOW_NAME, _env_=env, _run_name_=run_name, _wait_for_=[preprocess], _parameters_=parameters)

But it seems really weird that submit should have stopped working. Has anyone else ran into this issue?

I'm running a Scheduled flow that calls another flow using the

startFlowRun

method. Initially when I was calling the first flow without using the

wait

parameter, the flows worked fine, but after I added the parameter, the parent flow failed with

TypeError: Object of type SUCCESS is not JSON serializable

. I was wondering if this is a known issue. Prefect version is 1.4.0