Guys, I am currently facing an issue with using AWS + Prefect, wich boils down to a problematic query. I downloaded the source code of prefect and have been trying for some time to find this query to modify it for better performance. Below is the query in question: UPDATE deployment SET last_polled=$: : TIMESTAMP WITH TIME ZONE, status=$2: :deployment_status, updated=now() WHERE deployment.in IN (NULL) AND (1 !=1) or deployment.work_queue_id IN (3$: :UUID) If you could help me find and fix this query, I Would be grateful! The goal would be to remove the following part: “(NULL) AND (1!=1) OR”. Thank you in advance!

Hi need some help for comparing https://github.com/tensorlakeai/indexify vs prefect . Not feature wise . But on the core engine design wise

Hello there 👋, could someone please suggest a contact from the Prefect team (ideally UK-based) whom I could contact for a technical partnership? We are interested in adopting the Prefect solution.

@Bianca Hoch Just wanted to drop in to say I've been trying to troubleshoot a

prefect deploy

issue which was reliably producing badly built images for ECR which EKS then refused to run (EKS reported on bad hash on image pull, imagePullErr). I had had this issue for over a week and tried everything with installing/re-installing Docker, different package manager venvs, different project layouts for prefect.yaml, etc. In the end, the solution was to disable the containerd image store in Docker Desktop. With that checkbox unticked, the

prefect deploy

worked first time, with it checked it would reliably not work. This is Prefect 3.2 with Python 3.12.

Thank you. I am reading this.

was the copy/clone run option removed? I was using that for debugging small sections of my code that had failed without rerunning the whole flow!

Hey @Nate sorry to direct tag but are there any updates here (on being able to respect the prefect version in your lockfile)? Because this is happening again now https://app.prefect.cloud/account/2e97a726-e32b-48ab-90fe-8c542523240f/workspace/66cf6c8b-748e-4cb7-9448-c079d54e0e36/runs/flow-run/0685fcb9-f33e-7bca-8000-20ab00ec6ee1?entity_id=0685fcb9-f33e-7bca-8000-20ab00ec6ee1 It’s a pretty big blocker for us because we can’t currently trust that our jobs will run reliably.

Hi all, assume we have Prefect flows with long tasks and we want to support a cancel request that stops or kills "immediately" the flow and the running tasks. What is the best way to do so? Thanks! (From my understanding, I see that the default cancel flow didn't affect the current running tasks and they continue to work. Also, when using "set_task_run_state()" or "set_flow_run_state()" it is just labeling with "cancel" not really stopping the run.)

Hi guys, it is kind of pain to debug docker build errors in Prefect when built using

.deploy

, as Prefect does not output anything but the very latest error, which is mostly missing the context and the text of the real fatal error. While building (in the deploy step) a custom image, it only shows building, without the steps, and if an error happens, show the "error" message, while a the real message, when using docker build, is in the fatal message, which prefect just does not show. Can we make prefect show all of the building steps, just like docker build?

Anyone using Prefect along with Pulsar? Curious if any particulars issue you may have come across.

# UPDATE: check my reply for answer! Maybe this will help someone. Is there a way to rebuild the whole image with no cache? I am not able to find what I have to pass to

.deploy

to build the whole image again. I already tried

image=DockerImage(
            name="myworkflow",
            tag="latest",
            dockerfile="Dockerfile",
            build_kwargs={"nocache": True},

But for some reason the build_kwargs is not recognized (I am getting an error)

With the new Prefect Tier system we are trying to upgrade our

Legacy Free Tier

to

Starter Tier

, but when we try to initiate the process by clicking on "Upgrade" we get this error. I've already emailed

<mailto:help@prefect.io|help@prefect.io>

early on Friday - still waiting for answers. Does anyone know the normal response time? I haven't received a notification that the support case has been received. Is this a normal bug people have seen before?

Hey all, there probably is a way to set retention time for prefect server for logs etc? We already have 46 GB of the prefect server related database see

Hi guys, I'm trying to format logs at the end of my master flow using client.read_logs, and from that using flow_run_id and task_run_id to get flow and task names programmatically (and other information). Problem It seems, however, that on the UI (and within a python flow client) that I can't get all the logs of child flows within parent flows to be together. Only the tasks with the parent flow show logs and not subflows. This means I can't get ALL logs of a parent flow and then format it for like a 'success email' for the entire flow. Unless someone has a better idea that gives me an 'overall report' of a flow and all of it is subflows I am open to ideas here is my code

@task
async def check_logs():
    async with get_client() as client:
        # Ensure any pending logs are sent
        await APILogHandler.aflush()
        logs = await client.read_logs(
            LogFilter(flow_run_id={"any_": [runtime.flow_run.id]})
        )

        records = []
        for log in logs:
            
            # Gets task run and flow run info
            if log.task_run_id is not None:
                task_runs = await client.read_task_runs(
                    task_run_filter=TaskRunFilter(id=TaskRunFilterId(any_=[log.task_run_id]))
                )
                task_run = task_runs[0]
                task_run_name = task_run.name
                print("-------------------- Task Run Details ----------------------------")
                print(task_run)
                print("------------------------------------------------------------------")
                if task_run.flow_run_id is not None:
                    flow_runs = await client.read_flow_runs(
                        flow_run_filter=FlowRunFilter(id=FlowRunFilterId(any_=[task_run.flow_run_id]))
                    )
                    flow_run = flow_runs[0]
                    print("-------------------- Flow Run Details ----------------------------")
                    print(flow_run)
                    print("------------------------------------------------------------------")
                    if flow_run.flow_id is not None:
                        flow = await client.read_flow(flow_run.flow_id)
                        print(f"Flow name is {flow.name}")

            print(log)


@flow(log_prints=True)
def daily_flow():
    
    ## A subflow
    some_subflow()
    
    ## A task
    some_task()
    
    check_logs.submit()
    
    # only logs of the some_task are printed by the check_logs function, not the some_subflow function

Hi @Prefect I recently created my account using my work email and upgraded to the pro subscription. Then out of habit, I authenticated using my GitHub account, which has the same work email. It seems my account details have been mixed up now, my pro subscription is not showing in my account but my work pool is still showing. When I try and use the Prefect CLI its not communicating properly with the Prefect UI. Can someone please assist me? There is no support email to contact and all available time for booking with an engineer are outside my business hours here in Australia.

@Prefect Hi Iam trying to run prefext in my local environment but getting this error while starting the server

dt = <http://self.tz|self.tz>.convert(dt)

AttributeError: 'NoneType' object has no attribute 'convert'

@Marvin how can I add a tag to a deployment using Prefect api?

Hello everyone, We found that DaskTaskRunner is not necessarily needed anymore. We were wondering if we remove it, specifically for a Flow, how would it work? i.e. It will check the complexity and define the optimum threads? Here we refer to a common use:

...
@flow(
    name="Scrapers sub-flow",
    task_runner=DaskTaskRunner(
        cluster_kwargs={"n_workers": 1, "processes": False, "threads_per_worker": 20}
    ),
)
def scraper(splitted_files: list[str], run_timestamp: str, success_threshold: float):
    logger = get_run_logger() 
...

Q: What is a decent way to deal with a workflow that has a high number of really small tasks? The workflow makes about 40k http requests that I've represented as 40k tasks, in order to use prefects retry logic and have that representation of what succeeds and fails. Obviously this doesn't work too well from the UI side of things. It doesn't present itself well in the UI and it's probably a bit harder on the server than it needs to be. These tasks do go to the same worker and do (I believe) reuse the same http client sessions, so the http requests themselves are not really inefficient. The obvious optimization is to batch groups of requests into maybe 1000 requests per task or pick a tasking level that is a bit less verbose. The down side with batching requests is I'm not sure prefect has the concept of partial success or if there's a way to only retry requests that fail in a task ( ignoring any internal retry logic to a task). Ideally there would be a way to represent a high volume subtask where 39,500 success and 500 fail, retry only the 500 failed requests. I feel like the UI doesn't show this too well but it's pretty close. (Open source server). I'm using an iterator so it does at least hide the really long list by default. Any suggestions? Am I missing a feature of prefect that might handle this better?

@Marvin the endpoint patch /api/deployments/{id} returns {‘detail’: ‘Method Not Allowed’}, why?

@Jake Kaplan Do you have some planned upgrade for self-hosted UI?

hi all, is there a way to run child flows concurrently similar to task.submit?

Hey there, I'm reaching out because we would like to discuss the increase of a rate limit in our prefect cloud account. Already tried to reach out to the sales team but did not get a response yet. Any experiences how this can be approached? Thanks a lot in advance

Hi One of Prefect flows is being marked as Completed in the UI, while several tasks within the flow are still shown as Running. Could you help me understand what might cause this discrepancy and how to resolve it?

I used and promoted the use of Prefect in my previous company for over 3 years where we self hosted prefect-server in an AWS kubernetes namespace. Now I am at a new company and am trying to get IT to get prefect-server self hosted and deployed here. We have gcp and slightly different ways of handling certs and other bits. I am more like a flow developer and not a network k8s specialist. Also I was previously using prefect version 2 and not 3. The online docs don’t help all that much as they don’t cover enough cases. We are using the prefect-helm charts and have read the few paragraphs under the advanced tab in the prefect docs. Any other docs or advice I can read?

Hi everyone! I’m using this filter

task_filter = TaskRunFilter(
    tags=TaskRunFilterTags(all_=[tag]),
    state_type=TaskRunFilterStateType(
        any_=[
            StateType.RUNNING.value,
            StateType.PENDING.value,
            StateType.CANCELLING.value,
            StateType.SCHEDULED.value,
            StateType.PAUSED.value
        ]
    )
)

but despite in the class doc it is written that it returns task runs meeting all conditions, it looks like it is an OR. Does it happen to you also?

Hello prefect team. We are in the middle of upgrading from prefect 2 to 3. I'm having trouble getting the basic usage of a local prefect server working. I start it with:

prefect server start --port 4201

as I have before, but I get this message in the bottom right of the browser:

Can't connect to Server API at <http://localhost:4200/api>. Check that it's accessible from your machine.

and nothing shows up in the UI when I run flows. Can anyone help me figure out what is wrong? I have tried on each of 3.4.4, 3.4.5, 3.4.6, and 3.4.7. I can't get this working on any of them

Hey everyone, so I'm using the dask taskrunner to run a temporary cluster on AWS ECS (dask_cloudprovider.aws), I just wanted to ask if anyone knows if the scheduler/everything in dask automatically resorts to using a local cluster if it fails to connect to ECS? I've noticed that the logs show the dask scheduler and workers running on an ip address such as 127.0.0.1 which is the localhost. Yet I don't seem to notice any warning message indicating that it can't connect to AWS

Hello prefect team! I’m playing with custom event automation on 3.2.11 but can’t seems to get it to trigger 😢 , here’s my automation trigger json

{
  "type": "event",
  "match": {},
  "match_related": {},
  "after": [
    "some.event.completed"
  ],
  "expect": [
    "some.event.completed"
  ],
  "for_each": [],
  "posture": "Reactive",
  "threshold": 1,
  "within": 3600
}

And here’s an event I sent and confirmed the engine received, does anyone know why won’t this event trigger the automation?

Is the Teams functionality broken in marvin 3.1.2 ? I have not had a successful implementation of the Teams functionality following the example documentation here. https://askmarvin.ai/concepts/agents