Hey I believe that prefect is still not traversing the path when trying to pull a flow. It is cloning the repo fine and I can see the repo in the docker container but it does not find the flow.

Has anyone seen this behavior on local Prefect with interval schedules? this deployment cli flag set:

--interval 600 --timezone America/Los_Angeles --anchor-date 2023-06-27T01:00:00Z

results in this schedule in UI Daily from Jun 26th, 2023 at 080000 PM (America/Los_Angeles)

I am in the course starting tomorrow. I got an error message, as follows, trying to install Prefect. (env1) C:\Users\maurreen.skowran>pip install -U prefect 'pip' is not recognized as an internal or external command, operable program or batch file.

Hello, I'm new to prefect and I'm using the python SDK to create my flows. I'm trying to get a list of all my flows deployed in my prefect UI environment and run them with their default parameters that I've set. Is there a call I can make that can get a list of all my flows and another call I can make that runs them by

name

or

id

?

Hi all; I’ve set up an instance of prefect server running in an on prem kubernetes cluster, and am trying to create a docker registry block. However, while creating the block requires a username and password, accessing our internal docker registry does not. What would I put in the username and password fields? And wouldn’t this same question arise when connecting to docker hub?

what is the recommended method for running sub flows sequentially where each subflow is a bunch of tasks mapped to run concurrently? i dont want the second subflow to run until the first one is complete. i tried the following but its complaining about the mapped list results… Example…

i get AttributeError: 'list' object has no attribute 'submit' because of the multiple tasks

@task
def task1(item: str):
    # make api call; save data
    return featherfilepath

@task 
def task2(item: str):
    # process stats; save data
    return featherfilepath

@flow # flowrun task1
def map_task1():
    # get list of items to process; convert to list
    items = feather_instance.get("items")
    items = items['name'].tolist()
    # map a task to each item in the list; run concurrently
    task1_multiple = task1.map(item=items)

@flow # flowrun task2
def map_task2():
    # get list of items to process; convert to list
    items = feather_instance.get("items")
    items = items['name'].tolist()
    # map a task to each item in the list; run concurrently
    task2_multiple = task2.map(item=items)

@flow
def flowrun():
    # api calls; cache from feather for one hour
    task1 = map_task1().submit()
    # process ta after all klines are saved to feathers
    task2 = map_task2().submit(wait_for=[task1])

if __name__ == "__main__":
    flowrun()

^ i think i figured it out. do this look correct? (also removed .submit())

@flow(task_runner=ConcurrentTaskRunner())  
def map_task1():

@flow(task_runner=ConcurrentTaskRunner())  
def map_task2():

@flow(task_runner=SequentialTaskRunner())
def flowrun():

Hi all. I’ve been facing an issue where my scheduled flows on Prefect cloud crash with no logs and this state message:

Flow run could not be submitted to infrastructure

. My Prefect is using ECS as infra basically following this (I’m new to using Terraform & ECS btw). We’ve been using this set up since the beginning of the year but the crashes started happening out of nowhere ~16 hours ago. Any advice welcome.

In the GUI for Prefect 2, is there a way to see the number of subtasks from a Mapped task that are Completed vs Running vs Pending?

Hi all, I am doing the migration from Prefect 1 to Prefect 2. Context: - Using Prefect Cloud - Using GCP (Prefect Agent running on VM, Cloud Run for execution of flows) - Github as storage I have gotten everything running in a production like environment. However I am now running into problems with downloading the src code before execution. I get the error:

prefect.exceptions.ScriptError: Script at 'src/flows/flow.py' encountered an exception: FileNotFoundError(2, 'No such file or directory')

We are using a monorepo where the prefect code sits in a subdirectory called “pipeline”. For deployment I am using the python SDK as such:

deployment = Deployment.build_from_flow(
        flow=flow,
        name=target_env,
        version=3,
        path="pipeline",
        work_queue_name=target_env,
        tags=[target_env],
        infrastructure=infrastructure_block,
        storage=storage_block,
    )

As mentioned it was working in my other environment when I did not use a subdirectory and did not provide a path. I have searched thoroughly in discourse, Slack, documentation and in the source code it self, but am running out of options. Does anyone see an error with the deployment or have suggestions on resources I can follow to get it working?

Hey guys , I recently moved my project to a new VM on E2E and , on typing prefect deployment ls , i get this error

Traceback (most recent call last):
  File "/home/cubeserver/contact-management-notifications/env/lib/python3.10/site-packages/prefect/cli/_utilities.py", line 41, in wrapper
    return fn(*args, **kwargs)
  File "/home/cubeserver/contact-management-notifications/env/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 255, in coroutine_wrapper
    return call()
  File "/home/cubeserver/contact-management-notifications/env/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 383, in __call__
    return self.result()
  File "/home/cubeserver/contact-management-notifications/env/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 283, in result
    return self.future.result(timeout=timeout)
  File "/home/cubeserver/contact-management-notifications/env/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 169, in result
    return self.__get_result()
  File "/usr/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/home/cubeserver/contact-management-notifications/env/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 346, in _run_async
    result = await coro
  File "/home/cubeserver/contact-management-notifications/env/lib/python3.10/site-packages/prefect/cli/deployment.py", line 460, in ls
    for deployment in sorted(
  File "/home/cubeserver/contact-management-notifications/env/lib/python3.10/site-packages/prefect/cli/deployment.py", line 449, in sort_by_name_keys
    return flows[d.flow_id].name, d.name
KeyError: UUID('aee72103-d63b-4e55-ae37-1fa2d19224c4')
An exception occurred.

Hi community, I have the following simple flow. This consists of only one load task.

@flow

def flow():

db_connection = db_connection()

with db_connection() as cursor:

load(cursor)

However, when I run it I get the error: Flow run encountered an exception. ProgrammingError: no results to fetch I have already found out that it is probably the cursor object which I pass to the task. Is there a way to make a database connection and then pass it to the tasks? thanks :)

Hello I was wondering if anyone have any advice on handling deployments locally vs production. I am currently using a github branch called prefect_test locally but in production I want to use main. Is there way of setting env variables to be use current branch locally and use main in production in the YAML?

pull:
    - prefect.deployments.steps.git_clone:
        repository: <https://github.com/gaia-family/monorepo.git>
        branch: prefect_test
        access_token: "{{ prefect.blocks.secret.github-access-token }}"
    - prefect.deployments.steps.set_working_directory:
        id: change_directory
        directory: /monorepo/data/prefect
    - prefect.deployments.steps.pip_install_requirements:
        requirements_file: requirements.txt

For workers and agents running on EC2, ECS etc.. is there any metadata saved to the flow_run regarding which machine or task_id it was run on?

Hey, I am trying to run my workflow with prefect in docker. The package is added in the requirements.txt and I have also added the following lines

RUN mkdir /.prefect  && chown -R random_program_planner /.prefect

USER random_program_planner

However I get the following error when I try to trigger the workflow from docker

/usr/local/lib/python3.9/site-packages/prefect/context.py:505: UserWarning: Failed to create the Prefect home directory at /home/random_program_planner/.prefect

Any help is appreciated 🙂 Thanks

Agent health question What cause agent/work queue to stop being healthy? Is the only thing to do when a queue becomes unhealthy is running

prefect agent start --pool default-agent-pool

and

prefect agent start -q <name>

? or is there are other things I can do to understand what is going on and prevent from it happening ?

If I do the

Create Flow Run

server API call and don't pass any parameters in. Will it just used the default parameters set in the deployment?

hey guys, in the deployment pull step can I get the access token from a prefect block to authenticate with gitlab? https://docs.prefect.io/2.10.18/concepts/deployments-ux/#the-pull-action:~:text=Below%20is%20an%20exa[…]requirements.txt%20file%20after%20cloning%3A

one more question, wondering where the default location the git repo is cloned to in the execution environment

my deployment is stuck at cloning the repo

it's been in this step for 15min

image.png

nvm figured it out, it was because I needed to setup ssh

any simple way to delete/truncate all pending items from task runs? for example, i have an issue and stop the flow run or it crashes. now i have pending items in the postgres log preventing a new run from finishing.

This message was deleted.

Hello,i are using prefect 0.15.11 . It been more 1.5 years we have been using but recently i am facing issue with prefect server that crash very frequently i don’t find any pattern in server crash and sometime agent get disconnected. In Server logs found following error . Can help me out what is the route cause of it.

{"severity": "ERROR", "name": "prefect-server.Scheduler", "message": "Unexpected error: ReadTimeout(TimeoutError())", "exc_info": "Traceback (most recent call last):
  File "/prefect-server/src/prefect_server/services/loop_service.py", line 60, in run
    await self.run_once()
  File "/prefect-server/src/prefect_server/services/towel/scheduler.py", line 47, in run_once
    offset=500 * iterations,
  File "/prefect-server/src/prefect_server/database/orm.py", line 501, in get
    as_box=not apply_schema,
  File "/prefect-server/src/prefect_server/database/hasura.py", line 85, in execute
    as_box=as_box,
  File "/prefect-server/src/prefect_server/utilities/graphql.py", line 64, in execute
    timeout=30,
  File "/usr/local/lib/python3.7/site-packages/httpx/_client.py", line 1385, in post
    timeout=timeout,
  File "/usr/local/lib/python3.7/site-packages/httpx/_client.py", line 1148, in request
    request, auth=auth, allow_redirects=allow_redirects, timeout=timeout,
  File "/usr/local/lib/python3.7/site-packages/httpx/_client.py", line 1169, in send
    request, auth=auth, timeout=timeout, allow_redirects=allow_redirects,
  File "/usr/local/lib/python3.7/site-packages/httpx/_client.py", line 1196, in send_handling_redirects
    request, auth=auth, timeout=timeout, history=history
  File "/usr/local/lib/python3.7/site-packages/httpx/_client.py", line 1232, in send_handling_auth
    response = await self.send_single_request(request, timeout)
  File "/usr/local/lib/python3.7/site-packages/httpx/_client.py", line 1269, in send_single_request
    timeout=timeout.as_dict(),
  File "/usr/local/lib/python3.7/site-packages/httpcore/_async/connection_pool.py", line 153, in request
    method, url, headers=headers, stream=stream, timeout=timeout
  File "/usr/local/lib/python3.7/site-packages/httpcore/_async/connection.py", line 78, in request
    return await self.connection.request(method, url, headers, stream, timeout)
  File "/usr/local/lib/python3.7/site-packages/httpcore/_async/http11.py", line 62, in request
    ) = await self._receive_response(timeout)
  File "/usr/local/lib/python3.7/site-packages/httpcore/_async/http11.py", line 115, in _receive_response
    event = await self._receive_event(timeout)
  File "/usr/local/lib/python3.7/site-packages/httpcore/_async/http11.py", line 145, in _receive_event
    data = await self.socket.read(self.READ_NUM_BYTES, timeout)
  File "/usr/local/lib/python3.7/site-packages/httpcore/_backends/asyncio.py", line 135, in read
    self.stream_reader.read(n), timeout.get("read")
  File "/usr/local/lib/python3.7/contextlib.py", line 130, in __exit__
    self.gen.throw(type, value, traceback)
  File "/usr/local/lib/python3.7/site-packages/httpcore/_exceptions.py", line 12, in map_exceptions
    raise to_exc(exc) from None
httpcore._exceptions.ReadTimeout"}

As per above error it shows it was due graphql connection.pls can someone suggest solution for this

How do I specify which subnet and security group should be applied to ECS Tasks spun up by an ECS Worker? In Prefect 1 we were able to include that in the definition for our ECS Agent, but those settings aren’t being applied to ECS tasks spun up by our Prefect 2 Worker. We’re using Terraform to deploy the Worker (and were using Terraform for our Prefect 1 ECS Agent too)

in the prefect.yaml file I specify the pull step to be a git clone, if I try deploying more than once, on the second time I get an error because the repo already exists in the working directory. Is there a way to specify to not clone but do a git pull from remote branch if the repo already exists?

Can I create triggers in the definitions section of the

prefect.yaml

file and reuse the same trigger across multiple deployments? Or do I have to write out the trigger for each deployment?

Hey team, I am trying to stop the local runs from logging in the prefect cloud UI as it makes monitoring our real jobs more difficult. The following in not stopping the logs from being written, even though printing

PREFECT_LOGGING_TO_API_ENABLED.value()

is showing that the value is false. Any thoughts?

with temporary_settings(updates={PREFECT_LOGGING_TO_API_ENABLED: False}):
     run()