Hi Everyone! I am new to Prefect and I tried following the getting started guide. Got a`sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) database is locked` error through the code here https://www.prefect.io/guide/blog/getting-started-prefect-2/. Is this a bug?

Hi Team, can anyone please help me with the significance of this error

06:55:16.136 | ERROR   | prefect.infrastructure.process - Process '<function uuid4 at 0x7fe62066ab80>' exited with status code: -9; This indicates that the process exited due to a SIGKILL signal. Typically, this is either caused by manual cancellation or high memory usage causing the operating system to terminate the process.

Hi all, start prefect2 server failed. Can anyone help? Thanks!

File "/Users/shelocks/.pyenv/versions/3.9.0/lib/python3.9/site-packages/prefect/server/api/server.py", line 502, in stop_services
    if app.state.services:
  File "/Users/shelocks/.pyenv/versions/3.9.0/lib/python3.9/site-packages/starlette/datastructures.py", line 705, in __getattr__
    raise AttributeError(message.format(self.__class__.__name__, key))
AttributeError: 'State' object has no attribute 'services'

Application startup failed. Exiting.
Server stopped!

Hello, I thought that exceptions are not caches. But I see this exception being cached [exception from httpx]

Is it possible to submit a flow-run from a deployment to be run anytime in the future? Want to check the status of flow-run asynchronously from my code.

Hi, all. My queue is unhealthy, and the amount of late flow runs is TOO BIG How can i delete all flow runs that are late using the web ui fast? If i did not delete those, i can not start the agent and make the work queue healthy.

Hello. I submitted 20 tasks to be executed with a concurrency limit of 4 (using tags). This results in no tasks being executed. All tasks are pending. Workers just spam this:

prefect-prefect-agent-3   | 11:17:06.370 | DEBUG   | prefect.engine - Received wait instruction for 30s: Concurrency limit for the max_4_concurrent tag has been reached

What am I doing wrong? For context: each task has to fetch a thing from the databse, send a POST request to an ML model, write a thing to the databse. I need no more than 4 tasks being executed in parllel, so that the model and the database are not overwhelmed.

Also I made a lot of tasks and three workers, but only one worker is executing the tasks. How can I have this pattern: a work queue is filled with tasks to do, multiple workers do the tasks in parallel? I thought it was a very basic case for a tool like this

hi all! running prefect server and agent (and db) in a docker. Getting the following message from the agent:

prefect-agent-1     | Failed the last 3 attempts. Please check your environment and configuration.
prefect-agent-1     | Examples of recent errors:
prefect-agent-1     | 
prefect-agent-1     | Traceback (most recent call last):
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/httpx/_transports/default.py", 
prefect-agent-1     | line 60, in map_httpcore_exceptions
prefect-agent-1     |     yield
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/httpx/_transports/default.py", 
prefect-agent-1     | line 353, in handle_async_request
prefect-agent-1     |     resp = await self._pool.handle_async_request(req)
prefect-agent-1     |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
prefect-agent-1     |   File 
prefect-agent-1     | "/usr/local/lib/python3.11/site-packages/httpcore/_async/connection_pool.py", 
prefect-agent-1     | line 208, in handle_async_request
prefect-agent-1     |     raise UnsupportedProtocol(
prefect-agent-1     | httpcore.UnsupportedProtocol: Request URL is missing an 'http://' or 'https://' 
prefect-agent-1     | protocol.
prefect-agent-1     | 
prefect-agent-1     | The above exception was the direct cause of the following exception:
prefect-agent-1     | 
prefect-agent-1     | Traceback (most recent call last):
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/prefect/utilities/services.py", 
prefect-agent-1     | line 53, in critical_service_loop
prefect-agent-1     |     await workload()
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/prefect/agent.py", line 264, in 
prefect-agent-1     | check_for_cancelled_flow_runs
prefect-agent-1     |     named_cancelling_flow_runs = await self.client.read_flow_runs(
prefect-agent-1     |                                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
prefect-agent-1     |   File 
prefect-agent-1     | "/usr/local/lib/python3.11/site-packages/prefect/client/orchestration.py", line 
prefect-agent-1     | 1700, in read_flow_runs
prefect-agent-1     |     response = await <http://self._client.post|self._client.post>("/flow_runs/filter", json=body)
prefect-agent-1     |                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1845, in
prefect-agent-1     | post
prefect-agent-1     |     return await self.request(
prefect-agent-1     |            ^^^^^^^^^^^^^^^^^^^
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1530, in
prefect-agent-1     | request
prefect-agent-1     |     return await self.send(request, auth=auth, 
prefect-agent-1     | follow_redirects=follow_redirects)
prefect-agent-1     |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
prefect-agent-1     | ^
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/prefect/client/base.py", line 
prefect-agent-1     | 251, in send
prefect-agent-1     |     response = await self._send_with_retry(
prefect-agent-1     |                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/prefect/client/base.py", line 
prefect-agent-1     | 194, in _send_with_retry
prefect-agent-1     |     response = await request()
prefect-agent-1     |                ^^^^^^^^^^^^^^^
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1617, in
prefect-agent-1     | send
prefect-agent-1     |     response = await self._send_handling_auth(
prefect-agent-1     |                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1645, in
prefect-agent-1     | _send_handling_auth
prefect-agent-1     |     response = await self._send_handling_redirects(
prefect-agent-1     |                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1682, in
prefect-agent-1     | _send_handling_redirects
prefect-agent-1     |     response = await self._send_single_request(request)
prefect-agent-1     |                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1719, in
prefect-agent-1     | _send_single_request
prefect-agent-1     |     response = await transport.handle_async_request(request)
prefect-agent-1     |                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/httpx/_transports/default.py", 
prefect-agent-1     | line 352, in handle_async_request
prefect-agent-1     |     with map_httpcore_exceptions():
prefect-agent-1     |   File "/usr/local/lib/python3.11/contextlib.py", line 155, in __exit__
prefect-agent-1     |     self.gen.throw(typ, value, traceback)
prefect-agent-1     |   File "/usr/local/lib/python3.11/site-packages/httpx/_transports/default.py", 
prefect-agent-1     | line 77, in map_httpcore_exceptions
prefect-agent-1     |     raise mapped_exc(message) from exc
prefect-agent-1     | httpx.UnsupportedProtocol: Request URL is missing an 'http://' or 'https://' 
prefect-agent-1     | protocol.
prefect-agent-1     | 
prefect-agent-1     | 
prefect-agent-1     | Backing off due to consecutive errors, using increased interval of  60.0s.

Has anyone seen something similar?

Hi there, quick question here : what's the difference between

s3

from

prefect.filesystems

and

S3Bucket

from

prefect_aws.s3

?

Hi all! Has anyone seen this:

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 879, in exec_module
  File "<frozen importlib._bootstrap_external>", line 1016, in get_code
  File "<frozen importlib._bootstrap_external>", line 1073, in get_data
FileNotFoundError: [Errno 2] No such file or directory: '/tmp/tmp832ebukxprefect/flow.py'

Hello, question:

# 1 - run foo in a flow, result cached for 10s
@task(cache_key_fn=task_input_hash, cache_expiration=pendulum.duration(seconds=10))
def foo():
    pass

# 2 - after 10s, run foo in a flow, result cached for 10h
@task(cache_key_fn=task_input_hash, cache_expiration=pendulum.duration(hours=10))
def foo():
    pass

# 3 - run foo in a flow, result still cached for 10h (same code and inputs as before)
@task(cache_key_fn=task_input_hash, cache_expiration=pendulum.duration(seconds=10))
def foo():
    pass

Any way to change

cache_expiration

?

refresh_cache

only updates the result in cache but does not reset

cache_expiration

. Reason: have tasks with large cache_expiration (hours/days) but if needed reset the cache (update the result) and restart the expiration from that moment.

Hi! I'm trying to limit tasks concurrent access to a resource using

anyio.CapacityLimiter

but after some tasks finish the flow seems to hang and no more tasks execute, can you help me? (code example and logs in thread)

When using the new Projects feature does applying the deployments also apply the definitions or do I need to also deploy those via the traditional way with either python or in the ui?

is anybody else having issues with

PREFECT_FLOW_DEFAULT_RETRY_DELAY_SECONDS

? doesn't seem to work from both ui or config, keeps setting it to 0seconds

is it possible to access

runtime.deployment

information when setting a Flow's run name? here's an example:

from prefect import flow, runtime
import datetime

def get_flow_run_name():
    date = datetime.datetime.now().isoformat()
    return f"{runtime.deployment.name}-{date}"

@flow(log_prints=True, flow_run_name=get_flow_run_name)
def hi_named_flow(name: str):
    print(f"Hi {name}")

when I try to deploy this with local Prefect server, agent, and storage, the flow run hangs here:

11:32:06.748 | INFO    | Flow run 'warm-numbat' - Downloading flow code from storage at '/<some_path>/flows/test'

but if use

runtime.flow_run.flow_name

instead of

runtime.deployment.name

, it works

Hi all. I'm using Kubernetes, and for example I have "large" flows that use a memory limit of 10Gi, and my Kube namespace quota will only allow one of these to run at a time (without exceeding the quota memory limit). So I have a concurrency limit in Prefect set up to only allow one "large" flow to run at once. Still, it seems as if sometimes Prefect is starting up a new "large" pod while another is still running, so Kube kills the pod with a quota exceeded error. Is this because of the default

prefetchSeconds: 10

setting? Do I need to set prefetchSeconds to 0 for the agent?

Hello, Is there anywhere I can find more examples on creating deployments from python object?

Trying to set up a CI/CD pipeline that registers flows. It just repeats this

18:44:38.783 | INFO    | botocore.credentials - Found credentials in environment variables.

and eventually times out. I'm running the

py

file that has the Flow as a script, and have this at the end:

if __name__ == "__main__":
    for memory in [15360, 30720, 63398, 126899]:
        base_args = make_deployment_args(flow_name1, memory, user="matt")
        storage1 = S3Bucket.load(<name>)
        deployment1 = Deployment.build_from_flow(
            flow=nb_flow, storage=storage1, **base_args
        )
    memory2 = 14360
    base_args2 = make_deployment_args(flow_name2, memory2, user="matt")
    storage2 = S3Bucket.load(<name>)
    deployment1 = Deployment.build_from_flow(
        flow=nb_subflows, storage=storage2, **base_args2
    )

It works fine when I run it interactively, but not in CI/CD.

Does anyone have suggestions on how to run two subflows simultaneously as their own ECS tasks? I have a diagram below of what I'm running currently running in Prefect 1. I want to replicate it in Prefect 2 After we complete the QA flow we call the Aggregations and Dump Raw flows, which run at the same time. They don't interact with each other at all - totally different tables etc.

image.png

My big stumbling point I think, is that I either need to combine both sequential and parallel executions aggregator and campaign statistics have to run sequentially, and dump_raw should run in parallel to both of those. That makes me think that there are two ways to go about this - 1) aggregator should be a flow that calls campaign_statistics as a task, and dump_raw is either a flow or a task that runs parallel 2) All three processes are tasks, and I call (aggregator then campaign statistics) in a sequential task runner which is itself inside a parallel task runner with dump_raw. I'm looking at the recipe here, which seems promising - https://github.com/PrefectHQ/prefect-recipes/blob/main/flows-advanced/parent-orchestrator/pokemon_weight.py

I am using the projects flow and building and docker image with the code baked into it. How do I tell the deploy to use the correct path as it keep trying to use the path of the code on my local machine and no the path in the container

I have a fairly hilarious problem that I’m not quite sure how to solve. So I’m running self-hosted Prefect, and the UI had become ungodly slow - taking 15-20 seconds or more to resolve sometimes. No worries, still useable, new product, etc, etc Went in to see if it was something I could put in a bugfix for, and I discovered the UI is trying to download about 1GB of data when trying to render the page lol [-> 🧵 ]

Does anyone know if it's possible to run a flow with dynamic tags, something like this:

tag1 = "mytag1"
tag2 = "mytag2"
with tags(f"{tag1}, {tag2}"):
        state = my_flow()

Or maybe another way to do it?

I'm trying out projects to manage prefect deployments. I tried using the docker-git recipe to initialize my project and then also set up an image to be built and pushed to a private registry with authentication with a docker-registry-credentials block; however, when I run the deployment, I get

"docker.errors.APIError: 500 Server Error for <http+docker://localhost/v1.43/images/create?tag=develop&fromImage>: ...... hiding this part of the URL..... : Internal Server Error ("Head "<https://hidingregistryurl/prefect/prefect-project/job1/manifests/develop>": denied: access forbidden".

How do I provide to my deployment an image pull secret? This is how my deployment is configured at the moment:

deployments:
- name: job1
  entrypoint: ./job1/flow.py:test_flow
  work_pool:
    name: example-project
    job_variables:
      image: '{{ image_name }}'

Another question: my team is working on creating a git-based development workflow using a self-hosted prefect server on our kubernetes cluster. Due to the sheer number of different "projects" or "jobs" we want to group multiple prefect projects into one repo. Here's an example structure:

git-repo-name/
      project1/
           flow.py
           requirements.txt
      project2/
           flow.py
           requirements.txt

Would it be best to initialize each project subdirectory as a prefect project with it's own prefect.yaml and deployment.yaml, or have it in the root directory and configure different deployments to use the different subdirectories. We also want to isolate dependencies for each project subdirectory in it's own container, so this will also have to influence the architecture of this development workflow. The configuration for each project subdirectory's deployment will be fairly similar, so what would you all suggest as the cleanest way to initialize for example a project 3? Is there some way to make "custom recipes" that can be used in prefect project init --recipe recipeName or is there a different solution such as using yaml aliases

Hello, I am using this video to create a ECS Cluster with prefect. It pulls from Prefect Image

prefecthq/prefect:2-python3.10

but I want to run a dbt command and dbt sits in another Repo, in ECR

<http://999999999999.dkr.ecr.us-east-1.amazonaws.com/analytics_dbt:latestI|999999999999.dkr.ecr.us-east-1.amazonaws.com/analytics_dbt:latestI>

I want to understand, how I can leverage the Prefect agent and execute dbt in another container? Should it lunch 2 containers?

Is there anyway to use a custom package that is installed in my environment to declare where the flow function is? Currently the deployment object needs a path, and does not accept the flow definition without a path.

Hello, I wanted to ask if there is an opportunity to specify the execution of commands in the terminal as a task ? I tried to create a function inside which I just call the command I need using python.

@task(name="synchronise data")
def sync_data() -> None:
    """Sync data to remote server"""
    print("Start synchronisation")
    rsync_cmd = (
        f"rsync "
        f"-avzx "
        f"--no-perms "
        f"--progress "
        f"{my_data_folder} "
        f"{my_server}:{servers_sync_folder}"
    )

    return None

It works great and logs, but I have no way of knowing if this task has failed with an error. Or not. I can find out about this only by looking through the logs.

rsync: [sender] write error: Broken pipe (32)

But python itself believes that the command has completed successfully )