prefect-community #prefect-community

Hey I have a nested map (three) It is using the output of the last result, instead of using the list of outputs. Do i need to do something to make the output of the 1st map iterable?

Florian Kühnlenz

05/25/2022, 7:06 PM

Are there problems with accessing prefect secrets right now?

John O'Farrell

05/25/2022, 7:31 PM

Hello, I'm currently trying to register a flow to amazon ECR using the docker storage. I want to include an existing dockerfile to handle the different modules the flow needs, but when I add the dockerfile parameter when setting the flow's storage it causes the process to get stuck on

prefect.Docker:Building the flow's Docker storage...

after creating a new temp directory with the dockerfile and

healthcheck.py

Jared Noynaert

05/25/2022, 7:34 PM

@Kevin Kho, did you ever find out more about https://github.com/PrefectHQ/prefect/issues/5769? This is impacting us on AKS also edit: the reason I'm asking is that we are probably going to have to change our default executor for all flows, just to enable task observability....!

Nikhil Jain

05/26/2022, 12:08 AM

@Kevin Kho I am unable to access SSM parameters and ENV variables in my flow task. I am using

ECSRun

as follows:

Copy code

ecs_run = ECSRun(
    run_task_kwargs={'cluster': 'dev-prefect-cluster'},
    task_role_arn='arn:aws:iam::<>:role/dev-task-runner-ecs-task-role',
    execution_role_arn='arn:aws:iam::<>:role/dev-task-runner-ecs-task-execution-role',
    image='<>.<http://dkr.ecr.us-west-1.amazonaws.com/dev-automation-scripts-ecr:latest|dkr.ecr.us-west-1.amazonaws.com/dev-automation-scripts-ecr:latest>',
    labels=['ecs', 'dev']
)
flow.run_config = ecs_run
...

And then in my terraform config for the

prefect-agent

, I am setting a

task_definition

. And this

task_definition_file

is stored in s3 and contains a list of

env

parameters and

secrets

that are pulled from SSM parameter store. What’s happening is that when the flow-task is created, it is not getting any of the

env

and

secrets

that I supply in the task_definition_file. I can kinda see why this is happening: I am not supplying any task_definition in the

ECSRun

config during flow registration. The reason is 2-fold: I don’t want to hard-code these env/SSM parameters in python code, they are supposed to live in terraform configs. And also, we want to share these env and ssm parameters across many tasks, so don’t want to have to define them everytime. Is there a way to accomplish this?

Alex de Geofroy

05/26/2022, 1:13 AM

Hey everyone. Not sure the right person to address this to, but I'm having issues with creating Prefect Agents in Azure. The error I'm receiving is "This offer is not available for subscriptions from Microsoft Azure Cloud Solution Providers." Is this something the Prefect team can enable, or is there other guidance on how to configure a subscription that will enable me to use the Agent? Here's an article I found describing the process for enabling the CSP program: https://docs.microsoft.com/en-us/azure/marketplace/cloud-solution-providers?WT.mc_id=AZ-MVP-5000120

Thomas Henn

05/26/2022, 5:25 AM

Hi everyone ! What is the proper way to use a debugger with Prefect Orion? For example, is there any debug setting so that Prefect Orion doesn't capture the exceptions within tasks? I couldn't find anything about this in the documentation. Thanks in advance!

Dharit Sura

05/26/2022, 5:55 AM

Hi everyone, I am having an issue in running Prefect 2.0 flows. Can anyone take a look at the error and help me understand what is the issue that is happening. I have tried to uninstall and reinstall prefect as well as python. But still the issue is continuing.

✅ 1

Dharit Sura

05/26/2022, 5:56 AM

✅ 1

Sumant Agnihotri

05/26/2022, 6:16 AM

Hi all, I have set up my Prefect Server, however, the

API Tokens

option is disabled. How do I connect to my server, to register Flows, etc?

Samuel Hinton

05/26/2022, 6:52 AM

Hi team! I currently have the slack integration turned on so I get flow failures posted. However, is there a way to actually configure what is reported using this web hook? Right now, every failure just gives us the same generic “Some reference tasks failed.” error message, when ideally it would be great to be able to pass on some of the logs from any task that failed. Is this possible to do?

Kayvan Shah

05/26/2022, 7:36 AM

What does this error trace mean?

Copy code

Traceback (most recent call last):
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/prefect/engine.py", line 468, in orchestrate_flow_run
    result = await run_sync_in_worker_thread(flow_call)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/prefect/utilities/asyncio.py", line 54, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(call, cancellable=True)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/home/kayvan/projects/prefect-demo/flows/spire_current_weather.py", line 55, in spire_current_weather
    sample = random.choice(conf)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/random.py", line 346, in choice
    return seq[self._randbelow(len(seq))]
TypeError: object of type 'PrefectFuture' has no len()

using random.choice() to select randomly from a list

Vadym Dytyniak

05/26/2022, 8:47 AM

Hi. Is it possible to access current flow object in custom task implementation?

Abokor Ahmed

05/26/2022, 8:50 AM

Hello team. I'm fairly new, forgive me for my stupidity. But I have this error: I want to run this command : " prefect create project tester " I have configure my config.toml but noting !

Shrikkanth

05/26/2022, 9:04 AM

Hi guys, I'm fairly new to Prefect and I'm trying to send reports of the state of the flow to slack. The below is my code and when I run it from Prefect cloud, I get the message in slack about the status but not able to get the error message in the message section. I need to know the function or where to call for editing the message section. The

<http://logger.info|logger.info>(lines)

contains the error message. Any suggestions will be helpful, thanks.

Sanjay Patel

05/26/2022, 10:32 AM

Hi, can I please get some assistance. I'm successfully running a flow in my dask cluster - the flow is launched from outside the dask cluster using the following

output = flow.run(executor=DaskExecutor(address=XXX, client_kwargs=XXX))

however the flow contains a task that actually has a prefect flow embedded in it. An unknown number of parallel tasks that the primary task prepares (lets say 15 of them) and then tries to execute. I would like the 15 to be done in parallel managed by dask. I don't want to create a new dask cluster from within the primary task (not even sure if that would work) but instead want to execute the flow in the same cluster. The following line works but i'm not sure it's going to utilize all the available workers and may only execute on the worker that initiates the flow. Any guidance on how I should be running the flow2 which is prepared and called the by the primary task in 'flow' above?

output2 = flow2.run()

I believe i'll need to use something like this if i was using pure dask - https://distributed.dask.org/en/stable/task-launch.html but i'm not sure the prefect method which will achieve this and would like to keep the additional features that prefect offers. Thanks so much in advance!

Stéphan Taljaard

05/26/2022, 10:45 AM

Hi. How do you save (and read) dollar signs in a secret? I'm on Prefect Server, implemented in GCP. The secret comes from Google Secret Manager. I have a custom secret function that, in development, pulls the secret from my PC's environment variables. I develop on Windows. I now have a DB with a password containing dollar signs. Because of this line, I can't read the password out correctly. Thoughts? Example:

Copy code

import os
from prefect.client import Secret

secret_name = 'JSON_SECRET'
# os.environ[
#     f"PREFECT__CONTEXT__SECRETS__{secret_name}"
# ] = '{"db1": {"password": "password", "user": "userA"}, "db2": {"password": "P@$$w0rd", "user": "userB"}}'

secret_value = Secret(secret_name).get()
print(secret_value['db2']['password'])
# This prints 'P@$w0rd', not 'P@$$w0rd'

Kayvan Shah

05/26/2022, 11:06 AM

Copy code

2022-05-26 11:04:12 | [INFO]:numexpr.utils - NumExpr defaulting to 2 threads.
Traceback (most recent call last):
  File "/home/kayvan/projects/prefect-demo/flows/spire_current_weather.py", line 65, in <module>
    spire_current_weather()
  File "/home/kayvan/projects/prefect-demo/flows/spire_current_weather.py", line 62, in spire_current_weather
    spire_pipe(obj, sample)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/prefect/flows.py", line 319, in __call__
    return enter_flow_run_engine_from_flow_call(self, parameters)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/prefect/engine.py", line 110, in enter_flow_run_engine_from_flow_call
    return anyio.run(begin_run)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/asyncio/base_events.py", line 647, in run_until_complete
    return future.result()
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/prefect/client.py", line 95, in with_injected_client
    return await fn(*args, **kwargs)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/prefect/engine.py", line 158, in create_then_begin_flow_run
    flow_run = await client.create_flow_run(
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/prefect/client.py", line 538, in create_flow_run
    response = await <http://self._client.post|self._client.post>("/flow_runs/", json=flow_run_create_json)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/httpx/_client.py", line 1820, in post
    return await self.request(
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/httpx/_client.py", line 1506, in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/prefect/client.py", line 233, in send
    response.raise_for_status()
  File "/home/kayvan/anaconda3/envs/prefect-conda/lib/python3.9/site-packages/httpx/_models.py", line 1510, in raise_for_status
    raise HTTPStatusError(message, request=request, response=self)
httpx.HTTPStatusError: Server error '500 Internal Server Error' for url '<https://api-beta.prefect.io/api/accounts/2693bb0e-7537-4327-83f2-06354b5d2b4d/workspaces/d8a21cf4-12f7-45f8-9323-5fe529e45148/flow_runs/>'
For more information check: <https://httpstatuses.com/500>

While running flow in flow

SULEMAN KHAN

05/26/2022, 12:10 PM

Hi, I am facing this problem. Issue:

Scheduled Flow

fails because task is unable to return value when

Prefect Server

runs the flow with

CloudFlowRunner

. Same Flow runs successfully when I use

flow.run()

. I have attached the code, Prefect UI screenshot, and terminal output of flow.run(). if-else block at lineno 920 in

task_runner.py

is causing issue. If I return a value from the

function/task

it runs the if-block otherwise else block. line

self.result.write(value, **formatting_kwargs)

is calling the function with format

self.

and dictionary

formatting_kwargs

also contains the

self

variable, this self variable contains the object of

Demo

class.

Rei Mendel

05/26/2022, 12:29 PM

Hey, how can I map flow of flows with dynamic parameters? something like this: (Is it possible to get the parameters from a task?)

Copy code

parameters = ['~/folder1', '~/folder2', '~/folder3']
    create_flow_run.map(flow_name="List dir", parameters={"folder_path": ???})

Luan Tiburcio

05/26/2022, 12:58 PM

Copy code

Good morning, I would like to know if there is a way to return a json directly through the graphql api in prefect1.0-ui??

Sang Young Noh

05/26/2022, 1:53 PM

Hello all I’m currently trying to run a code with a dask runner - I’ve set up a dask scheduler and client and tried running with something like the following code for the flow: I was just wondering if anyone was familiar with this error? Thanks

Jessica Smith

05/26/2022, 1:55 PM

can we get the logger name added to FlowRunLog? Not sure why it isn't there by default, it's kind of important

Stephen Lloyd

05/26/2022, 2:31 PM

I have a mapped task that will make an api call. What is the best way to count successful api responses? I could check the response message in the task and return an int along with the response. It seems like

send_tasks

will be a list of the responses and I could iterate over it in a subsequent task, but I’m hoping there is a more efficient method. Can I force a task to fail if the api indicates a failed call? If i do, is there a way to count those?

Copy code

@task
def make_post(creds, task):
   ...
   response = requests.request('POST', ...)
   return response

with Flow(stuff, stuff):
   send_tasks = make_post.map(creds=unmapped(creds), task=data)

Will Skelton

05/26/2022, 3:28 PM

Hi All! I'm working on getting a Linux VM in Azure to host a prefect 2.0 server for POC/Dev purposes. I've followed the tutorial I've linked here but have gotten to a roadblock when trying to connect to the UI from my local computer. I've enabled Firewall rules successfully enough to allow SSH access into the vm via port 22 and have also added ports 8080 and 4200 as described in the tutorial. My issue comes when trying to bind the server to the public IP address for the vm. I've added the config.toml file as shown in the tutorial and have also tried to bind the IP using the "prefect orion start --host" command. Any thoughts on what I'm missing? Thanks! Will https://mortimer.xyz/prefect-on-azure/

Matthew Seligson

05/26/2022, 4:11 PM

When using the GitLab storage, in what scenarios is the flow code pulled? Is there caching? My assumption is that flow code is pulled from gitlab only upon initial start and restart from a failed state. Can the flow code be pulled while some tasks are running?

Kyle McChesney

05/26/2022, 4:31 PM

Is there any documentation around ResourceManagers and concurrency. Mostly just wondering about something like

Copy code

with Flow(
    'resource_manager',
    executor=LocalDaskExecutor(),
) as flow:
    with ResourceManager() as manager:
        something_res = something(manager)
        get_res = another_thing.map(something_res, unmapped(manager))

Mostly wondering If I have the resource manager return “itself” how many copies are made via map

Edmondo Porcu

05/26/2022, 6:35 PM

Hi, are there examples of using https://docs.prefect.io/api/latest/tasks/aws.html#batchsubmit ?

Edmondo Porcu

05/26/2022, 6:37 PM

Also, if one uses a task that returns a list, can another task be executed for each element of the returned list?