All the files are pushed to s3 path specified in s3-bucket block as well

Hi all! I'm new to Prefect, and I'm trying to run a Flow and Tasks. I got the below error while running my python script.

RuntimeError: Cannot create flow run. Failed to reach API at <http://127.0.0.1:4200/api/>.

Any ideas on what the problem could be?

Hi all recently update all my ETLs to run in prefect 2.0 but now I am getting a problem with the agent crashing. I am not sure if it matters but I am running the agent in a virtual environment here are the error messages that I am getting Traceback (most recent call last): File "c:\etls2\lib\site-packages\prefect\utilities\services.py", line 46, in critical_service_loop await workload() File "c:\etls2\lib\site-packages\prefect\agent.py", line 194, in get_and_submit_flow_runs responses = await self.client.get_scheduled_flow_runs_for_work_pool( File "c:\etls2\lib\site-packages\prefect\client\orchestration.py", line 2282, in get_scheduled_flow_runs_for_work_pool response = await self._client.post( File "c:\etls2\lib\site-packages\httpx\_client.py", line 1845, in post return await self.request( File "c:\etls2\lib\site-packages\httpx\_client.py", line 1530, in request return await self.send(request, auth=auth, follow_redirects=follow_redirects) File "c:\etls2\lib\site-packages\prefect\client\base.py", line 249, in send response = await self._send_with_retry( File "c:\etls2\lib\site-packages\prefect\client\base.py", line 192, in _send_with_retry response = await request() File "c:\etls2\lib\site-packages\httpx\_client.py", line 1617, in send response = await self._send_handling_auth( File "c:\etls2\lib\site-packages\httpx\_client.py", line 1645, in _send_handling_auth response = await self._send_handling_redirects( File "c:\etls2\lib\site-packages\httpx\_client.py", line 1682, in _send_handling_redirects response = await self._send_single_request(request) File "c:\etls2\lib\site-packages\httpx\_client.py", line 1719, in _send_single_request response = await transport.handle_async_request(request) File "c:\etls2\lib\site-packages\httpx\_transports\default.py", line 352, in handle_async_request with map_httpcore_exceptions(): File "C:\Program Files\Python310\lib\contextlib.py", line 153, in exit self.gen.throw(typ, value, traceback) File "c:\etls2\lib\site-packages\httpx\_transports\default.py", line 77, in map_httpcore_exceptions raise mapped_exc(message) from exc httpx.RemoteProtocolError: <ConnectionTerminated error_code:ErrorCodes.NO_ERROR, last_stream_id:235, additional_data:73657373696f6e5f74696d65645f6f7574> Traceback (most recent call last): File "c:\etls2\lib\site-packages\prefect\cli\_utilities.py", line 41, in wrapper return fn(*args, **kwargs) File "c:\etls2\lib\site-packages\prefect\utilities\asyncutils.py", line 260, in coroutine_wrapper return call() File "c:\etls2\lib\site-packages\prefect\_internal\concurrency\calls.py", line 245, in call return self.result() File "c:\etls2\lib\site-packages\prefect\_internal\concurrency\calls.py", line 173, in result return self.future.result(timeout=timeout) File "C:\Program Files\Python310\lib\concurrent\futures\_base.py", line 439, in result return self.__get_result() File "C:\Program Files\Python310\lib\concurrent\futures\_base.py", line 391, in __get_result raise self._exception File "c:\etls2\lib\site-packages\prefect\_internal\concurrency\calls.py", line 218, in _run_async result = await coro File "c:\etls2\lib\site-packages\prefect\cli\agent.py", line 189, in start async with anyio.create_task_group() as tg: File "c:\etls2\lib\site-packages\anyio\_backends\_asyncio.py", line 662, in aexit raise exceptions[0] File "c:\etls2\lib\site-packages\prefect\utilities\services.py", line 104, in critical_service_loop raise RuntimeError("Service exceeded error threshold.") RuntimeError: Service exceeded error threshold.

Hello, it seems that timezone configuration at user level is not used for datetime parameters I'm migrating from Perefect 1 with built in DateTimeParameters to orion and it seems strange that all datetime parameters are naive now

Hi everyone, I want to interact with the prefect api from my django app, but the client for prefect2 is only async. Is there a synchronous client? what are the best ways to do this?

Hey Team, how long does prefect hold the flow runs for in the self hosted version of Prefect (in the UI and in the DB)? I see currently I have 330 ish flows that get cut off at around Apr 25.

Hi, i went to prefect cloud just now and it seems all the flows are gone, I still see the deployments, but the flows are gone, any issues happening?

Hello. In Prefect2, I have a flow inside which I run some deployments using

run_deployment()

. Now in my unit test, I want to mock

run_deployment

, but I don't know what to return. I tried returning a Future object, but didn't work because I do some

result.name, result.state_name

on the output of

run_deployment

. I appreciate any help here on how to construct a flow_run object for my run deployment mock. Thanks

Hi, I need help with Prefect 2.10.4. We have installed prefect locally and want to deploy flows with CI/CD pipeline through AWS lambda function with use of Prefect deployment SDK kit. When we reviewed SDK document we found there is no parameter do define URL or prefect server details. I need some help from community to understand if anyone did such setup before and how they are deploying flows with AWS lambda functions or through git repository. Alternate route we found as REST API but before going to that route need to know is their possibilities we have with SDK kit.

Need to be able to clone git over ssh in the pull step of a project, how do I go about setting that up? cant seem to find any docs detailing it

Hello everyone, I am creating a deployment whit this code:

def build_and_register_flow(schedule_time, flow_name):
    """Build the prefect flow."""
    response = run_deployment(
        name=flow_name,
        scheduled_time=schedule_time,
        idempotency_key=schedule_time.strftime("%Y-%m-%d %H:%M:%S"),
        timeout=0
        )
    print(response)

But I am getting:

<coroutine object run_deployment at 0x7fc5082b9f40>

And the deployment is not set. Does anyone know what is going on?

What’s the best way to get task/data dependencies for a flow via Prefect’s API?

Is there a good way to monitor agents in cloud 2.0 besides checking for runs that aren't picked up? Similar to how cloud 1 has the agents would be ideal. Having an issue with an agent running on kubernetes that loses connection to cloud periodically.

after updating prefect to the newest version I'm facing this error when deploying a flow:

'NoneType' object has no attribute 'get'

Anyone with that error too?

Hi, as of today I'm no longer seeing whether a flow is running, completed, pending etc in the tab icon in google chrome. I know 2.10.6 did something to the icons, was this intentional?

Hi everyone, I am getting the following error, but Finished Job Ttl is configured in Kubernetes Job block (tried via Finished Job Ttl input box and via Base Job Manifest). Log

Submission failed. kubernetes.client.exceptions.ApiException: (403) Reason: Forbidden HTTP response headers: HTTPHeaderDict({'Audit-Id': '8dafec7b-865c-4190-9ff8-2c6859310d44', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Kubernetes-Pf-Flowschema-Uid': '271f18eb-a208-418a-b558-589f2b1931ce', 'X-Kubernetes-Pf-Prioritylevel-Uid': '666d5261-3a03-4c64-b623-ed6529882f73', 'Date': 'Fri,
    28 Apr 2023 13: 47: 23 GMT', 'Content-Length': '332'
}) HTTP response body: {
    "kind": "Status",
    "apiVersion": "v1",
    "metadata": {},
    "status": "Failure",
    "message": "admission webhook \"validation.gatekeeper.sh\" denied the request: [job-ttl-after-finished-v2] Job does not have ttlSecondsAfterFinished configured",
    "reason": "[job-ttl-after-finished-v2] Job does not have ttlSecondsAfterFinished configured",
    "code": 403
}

Base Job Manifest

{
  "apiVersion": "batch/v1",
  "kind": "Job",
  "metadata": {
    "labels": {}
  },
  "spec": {
    "ttlSecondsAfterFinished": 120,
    "template": {
      "spec": {
        "serviceAccountName": "prefect-agent",
        "completions": 1,
        "containers": [
          {
            "env": [],
            "name": "prefect-job"
          }
        ],
        "parallelism": 1,
        "restartPolicy": "Never"
      }
    }
  }
}

Can anyone help me here? Thanks in advance!

Hi everyone, When i'm running my prefect agent i'm running into this issue

raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__  
prefect.exceptions.PrefectHTTPStatusError: Server error '503 Service Unavailable' for url 
For more information <https://httpstatuses.com/503>

and then my agent stops running is there a way to make sure my agent doesn't shut down? This has happened more than once and not sure if there's a a way to prevent this from happening?

Is there a way to filter flow runs by what is in their logs? For example I have a flow run that errored when trying to access a person by their ID, is there a way to try and find the flow run that created that person by searching for a flow run with the user ID present anywhere in the logs?

I did run

prefect server database reset -y

and now I'm unable to

prefect server start

I've been facing this error:

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) unable to open database file

Hey @Prefect: I'm having trouble getting the dependency arrows to show up (using Prefect 2.10.3 Self Hosted). I understand you can use the

wait_for

command when invoking a function to get the dependency arrows but I'm not seeing anything. Code snippet is below, along with UI output. Any help would be appreciated. Thanks!

from prefect import flow, task

@task
def task_1():
    pass

@task
def task_2():
    pass

@task
def task_3():
    pass

@flow
def my_flow():
    x = task_1()
    # task 2 will wait for task_1 to complete
    y = task_2(wait_for=[x])
    task_3(wait_for=[y])
    
if __name__ == "__main__":
   my_flow()

What’s the best way to persist artifacts (output files) from Prefect workflow runs in Kubernetes? I am running on AKS (Azure Kubernetes Service) and I would like to be able to access files generated during the run of the flows. I could mount a PersistentVolumeClaim into /my_data on the pod/container and have my workflow simply:

import os
PERSISTENT_OUT_DIR = '/my_data'
out_path = os.path.join(PERSISTENT_OUT_DIR, 'my_output.csv')
with open(out_path, 'w') as f:
  f.write('my,cool,csv,header')

And then

kubectl cp

or e.g. use a filebrowser to access them via a Web UI. However, I am sure that there are standard ways to upload generated files during the workflow runs from Prefect and I’d love to learn about the idiomatic approaches. I am guessing Prefect has support through its SDK to access S3 / Azure Blob Storage / Azure Files, etc. Your thoughts?

I can't get my work queues to start properly. Doing this on a RHEL VM with a service file that just calls

prefect agent start -q main

. When I look at the logs for this, it shows me the agent starts and that it's checking for runs but it doesn't find any. Meanwhile the work queue in Prefect Cloud is unhealthy with a bunch of late runs. I tried reauthenticating to Prefect Cloud from the terminal and restarting the agent and upgrading to the latest prefect version. Any other advice?

Is there a mechanism (existing or planned) to register custom project recipes within a given environment? I'm working in a restricted cluster with non-standard defaults and it would be nice if users could get those defaults on project init instead of the standard (for example) git pull steps that don't work in our environment.

I see that this page details a Flow Run Retention feature: https://orion-docs.prefect.io/2.7/ui/flow-runs/?h=retention#flow-run-retention-policy but it's only available for prefect cloud. We are self-hosting prefect but we want to implement something like this. Is it the case that if we delete a flow run, for example via the prefect server python client:

client.delete_flow_run(flow_run_id=flow_run.id)

Does that actually delete a) the logs b) any other data associated with that flow run? Also more, generally, is this the recommended approach for managing a growing database size as prefect flow runs accumulate?

Hi 👋 When running ECS Tasks, is there a way to take some identifier from the Flow’s run in Prefect’s dashboard, and find it in the AWS dashboard? Or alternatively, somehow modify the creation of the ECS Task so that it’s Id would be the same as the Prefect’s flow id.

Hi All - Relatively new here. Trying to work through a simple example of deploying a prefect project that uses s3 storage. Is it common for this method to take 30-60 minutes on a 2 file python project?

Hi all. Trying to get a Kubernetes agent to run flows with KubernetesJob infrastructure. Is there a way to customize what directory in the container running the Job is used as the working directory for downloading and running flow code?

Is there a way to pass arguments to the Docker build step of a project deployment?

Hi all, we are using airflow to trigger our data pipelines written in Java, we plan to use prefect. Is there a way that we can trigger the Java pipeline using prefect?