Does anyone have a small code snippet, example for logging to a database?

Hi Prefect devs, I've been working on a POC and got everything running with ECS and fargate. I have what might be a fairly unusual use case where I am scheduling flow runs (through the grqphql api) for a day in specific intervals to scrape certain data. These intervals are spread out and then as the event nears become closer (with flows running every 10 seconds as the event gets closer) (think of horse racing data as the race gets closer ). Anyway, with ECS scheduling close running jobs doesn't really work out as the start up time is for the task definition is ~ 45 seconds. A possible solution for this would be to give us the ability to run these jobs on Lambdas (they now support Docker image runs and we could keep the lambda warm for shorter start up times). Wondering if this is something that Prefect might consider in the future or is this just too much of a corner case and I'm missing the point on what Prefect should be used for?

Hey Community, I am launching some kubernetes pods using the RunNamespacedJob task and I am trying to get the logs from these task pods to show up on the Prefect UI (currently running a Prefect server instance on an Azure VM with in-cluster agents). I end up seeing the

Started following logs for <task pod>

and then nothing more other than the completion and deletion log. Meanwhile, if I check the kubernetes job pod logs (

kubectl logs -n my-namespace prefect-job-...

I see the logs of the task pod showing up nicely, just not being sent back to the the Agent/Server. Anyone have a similar experience and know of that one flag I'm missing?

hey is there a way to run a subset of a flow? ideally I could pick a single task in a flow and any dependent tasks for that specific flow would also be run

Hi everyone, I’m having trouble switching from Docker storage (which runs successfully) to Github storage.

flow.storage = GitHub(secrets=["GITHUB_ACCESS_TOKEN"],
                      repo="my_org/repo_name",
                      path="prefect_flows/flows/flow.py",
                      ref="trunk")

The access token is saved in prefect cloud under the same name. During the flow run prefect can’t seem to be able to find the repo??

INFO:Downloading flow from GitHub storage - repo: 'my_org/repo_name', path: 'prefect_flows/flows/flow.py', ref: 'trunk'

ERROR: Repo 'my_org/repo_name' not found. Check that it exists (and is spelled correctly), and that you have configured the proper credentials for accessing it.

ERROR: Failed to load and execute Flow's environment: UnknownObjectException(404, {'message': 'Not Found', 'documentation_url': '<https://docs.github.com/rest/reference/repos#get-a-repository>'}, {'server': '<http://GitHub.com|GitHub.com>', 'date': 'Fri, 11 Jun 2021 22:46:06 GMT', 'content-type': 'application/json; charset=utf-8', 'x-github-media-type': 'github.v3; format=json', 'access-control-expose-headers': 'ETag, Link, Location, Retry-After, X-GitHub-OTP, X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Used, X-RateLimit-Resource, X-RateLimit-Reset, X-OAuth-Scopes, X-Accepted-OAuth-Scopes, X-Poll-Interval, X-GitHub-Media-Type, Deprecation, Sunset', 'access-control-allow-origin': '*', 'strict-transport-security': 'max-age=31536000; includeSubdomains; preload', 'x-frame-options': 'deny', 'x-content-type-options': 'nosniff', 'x-xss-protection': '0', 'referrer-policy': 'origin-when-cross-origin, strict-origin-when-cross-origin', 'content-security-policy': "default-src 'none'", 'vary': 'Accept-Encoding, Accept, X-Requested-With', 'content-encoding': 'gzip', 'x-ratelimit-limit': '60', 'x-ratelimit-remaining': '59', 'x-ratelimit-reset': '1623455165', 'x-ratelimit-resource': 'core', 'x-ratelimit-used': '1', 'content-length': '112', 'x-github-request-id': 'C202:57D9:E1FC5:1D69CC:60C3E7AD'})

I’ve curled the github api endpoint

repos/

directly using the same auth token and I do get the repo’s information. Pls advise! 🙏

Hi everyone. Is there any way to write a general wrapper for the

Task.run

method? I'd like to do something like this:

class MyTask(prefect.tasks.core.function.FunctionTask):
    def run(*args, **kwargs):
        try:
            return self.super().run(*args, **kwargs)
        except BaseException as e:
            prefect.context.get('logger').error('Error on host {}: {}'.format(hostname, format_error(e)))
            raise # Or better yet, mark the task as Failed

By default, Prefect catches, logs, and marks the task as failed for some errors, but for others, it puts

str(e)

in the "State Message" and throws away the stack trace, and for still others, it drops the error completely and the task appears to be "running". My wrapper would fix these problems. But you can't write the method above because there's a function called

_validate_run_signature

that expressly forbids it. So the only way to do it that I can think of would be to either wait for Python to get macros, or to add code generation to my deployment process (it would insert the try/except block into the body of every function that has a

@task

decorator). Is it really impossible to write this wrapper as an ordinary method?

Error while deploying flow: ValidationError({'type': ['Unsupported value: UniversalRun']

with prefect version

0.14.21

Hello community!! I´m trying to spin up a K8s cluster using the helm chart provided by prefetc…amd I´m trying to use an external postgreSQL database. The thing is_ What should I have to write inside postgresql.externalHostname? the hole postresql url with user/password/url/port/ddbb ? or just the url?

💃 Just arrived!

hello, i've got a problem with running the ETL tutorial

specifically i'm getting this error

requests.exceptions.SSLError: HTTPSConnectionPool(host='<http://opensky-network.org|opensky-network.org>', port=443): Max retries exceeded with url: /api/states/all?lamin=37.15294209056667&lamax=40.75094679823332&lomin=-79.76136868215879&lomax=-75.13474242904122 (Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1129)')))

So in troubleshooting an error having to do with differences in python versions, I somehow broke what little I had working earlier today. Can anyone shed light on the following error:

ERROR - <name> | Failed to query for flow run metadata

Traceback (most recent call last):
  File "/home/ubuntu/.local/lib/python3.8/site-packages/prefect/agent/agent.py", line 324, in _submit_deploy_flow_run_jobs
    flow_runs = self._get_flow_run_metadata(flow_run_ids)
  File "/home/ubuntu/.local/lib/python3.8/site-packages/prefect/agent/agent.py", line 684, in _get_flow_run_metadata
    result = self.client.graphql(query)
  File "/home/ubuntu/.local/lib/python3.8/site-packages/prefect/client/client.py", line 319, in graphql
    raise ClientError(result["errors"])
prefect.utilities.exceptions.ClientError: [{'path': ['flow_run', 0, 'id'], 'message': 'Cannot return null for non-nullable field flow_run.id.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

I rebuilt my agent (was planning to anyway) but same error right away upon launching. I deleted and reinitialized my project. It doesn’t seem like I’m doing anything differently than before when the agent appeared happy but my issue was further down the line. Really mystified right now, but I’m a bit frazzled from working it so many hours…maybe I’m missing something obvious? (I’m also not the least bit experienced with Prefect - nor very much with AirFlow.)

does anybody have any doc to follow for deploying prefect in kubernetes cluster?

Is anyone else unsuccessful in storing their flows as scripts using

azure

storage? I had a look at comparing the

s3

implementation (https://github.com/PrefectHQ/prefect/blob/master/src/prefect/storage/s3.py#L159) to the

azure

one (https://github.com/PrefectHQ/prefect/blob/master/src/prefect/storage/azure.py#L128), and it seems like the

stored_as_script

flag doesnt do much on the

azure

storage. Can anyone from the Prefect team confirm what I am seeing or am I on the wrong track?

Hey, I have a question. Is it better to store flow as pickled object in S3 or store_as_script? What are drawbacks of one or other way?

HI All, When I delete project with prefect client API client.delete_project, The project is not deleted immediately. Has anyone experienced this before?

hi all, if i have an upstream prefect task submitting a job to aws batch and a downstream task which executes upon the aws batch job completes. is it good to have the downstream task to keep checking whether the aws batch job completes by using boto3 sdk? or is there a more elegant way to achieve this?

Don't suppose anyone is running a

prefect agent kubernetes

in a custom Docker Image? I can get my image running on AWS Fargate with:

command=[
    "prefect",
    "agent",
    "ecs",
    "start",
    "--agent-address",
    "http://:8080",
    "--cluster",
    cluster.cluster_arn,
    "--task-role-arn",
    ecs_task_role.role_arn,
],

But on AKS it's a whole different story, I can't get it running 😭 Essentially it can't find prefect but I can run the container locally and invoke prefect

hey guys - noticing some slow downs in prefect cloud today. Things that typically take <1 second (e.g. loading in flows to the flows view) are taking >5 in some cases. Just wanted to make you guys aware!

possibly a dumb question, but we just now are getting to the point of running things on schedules so I’m running into this. I want to schedule a flow so that it will run on a schedule in prefect cloud, however i still want it to run right away when flow.run is invoked. How can I do that? The schedule I’m using is below:

flow.schedule = CronSchedule("30 12 * * *")

I have noticed some unexpected behavior using the prefect AWSClientWait task on long running batch jobs. Sometimes the AWSClientWait task will have failed, and the schematic and run details show the task to have failed, but the flow state is hangs in the running state for hours after. This particular task is part of a mapped flow, and other mapped flows failed on the same task, yet their flow state is failed. Any idea why this might occur? The trigger rule for the task that follows the failed task is also all_successful so it shouldn't be waiting to submit another task

Hey all, I’m trying to set up an ECS agent to run code stored in GitHub. Right now, I’m able to run basic code, but all of the parameters set in ECSRun() seem to be ignored, like the docker image to use and the cpu/memory. I’ve seen this in a couple other threads but was unable to follow the solutions (something about overriding?) I was able to choose an image location by setting it in the Run mode in prefect cloud UI, but I had to clear the cpu and memory options or else it would give a “string not valid” error. I’m also wondering if there’s a better way to specify the image location w/out the UI?

Hello all, is there a standard way of migrating Prefect Server from one server to another? We will shortly be switching servers and would like to preserve historical run data and projects

Is there a way to add documentation to a Parameter that shows up in the UI?

Hi Prefect devs, is there a way in prefect to only increase version numbers of flows if there is a change detected? Thinking in terms of a ci/cd pipeline where all flows are registered as part of the deployment. Or is this logic something that is up to us as the users?

Hey everyone, is it possible to break a flow if a mapped task runs on error without executing the following mapped tasks? If not, any suggestions how to solve this problem? Szenario: Executing a pipeline for a list of files (1...n) but if one runs on failure the flow should stopp immediately. I tried it with state handlers and the flow runs on error. But all instances of the mapped task get executed before the flow runs on error.

Hello all

Hi, I'm having trouble with what looks like a Prefect Cloud API problem. I had a flow running which failed after an hour with the following error. The flow had 10+ tasks complete successfully first. I'm running it with Prefect Cloud and K8 agent. The flow have been running successfully before.

Failed to set task state with error: ClientError([{'path': ['set_task_run_states'], 'message': '[{\'extensions\': {\'path\': \'$.selectionSet.billing_license.selectionSet.account_name\', \'code\': \'validation-failed\'}, \'message\': \'field "account_name" not found in type: \\\'billing_license\\\'\'}]', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}])
Traceback (most recent call last):
  File "/root/miniconda3/lib/python3.8/site-packages/prefect/engine/cloud/task_runner.py", line 98, in call_runner_target_handlers
    state = self.client.set_task_run_state(
  File "/root/miniconda3/lib/python3.8/site-packages/prefect/client/client.py", line 1503, in set_task_run_state
    result = self.graphql(
  File "/root/miniconda3/lib/python3.8/site-packages/prefect/client/client.py", line 319, in graphql
    raise ClientError(result["errors"])
prefect.utilities.exceptions.ClientError: [{'path': ['set_task_run_states'], 'message': '[{\'extensions\': {\'path\': \'$.selectionSet.billing_license.selectionSet.account_name\', \'code\': \'validation-failed\'}, \'message\': \'field "account_name" not found in type: \\\'billing_license\\\'\'}]', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

I'm running Prefect 0.14.7 in the flows and the agent is 0.14.17

Can you please help me to send a eamil using EmailTask notification task?

Hi, can someone help me with this error? https://cloud.prefect.io/stockwell/flow-run/f82f5a87-e196-4019-a332-d27355fda2dd?logs=

KeyError: 'Task slug disable_sound_on_mixer_error_stores-1 not found in the current Flow; this is usually caused by changing the Flow without reregistering it with the Prefect API.'

The slug is a transformation of the name of a flow. Prefect replaced the hyphens with

_

and added

-1

to the end. The error is happening when trying to run a different flow. Both flows are defined in the same Python source file. Both flows were registered with the latest code, contrary to the suggestion in the error message.