prefect-community #prefect-community

Is there anything akin to a lambda/anonymous task in Prefect? E.g. I have an iterable and want to transform it into a dictionary. I could define a task to do so but in this case something that simple could be done with an anonymous function

Mitchell Bregman

02/10/2021, 7:31 PM

Is there a way to have 2 different schedule patterns attached to a single flow. For example, I want this flow to run on a 5 minute interval during business hours only, and on an hourly interval off business hours.

👍 1

Verun Rahimtoola

02/10/2021, 7:40 PM

hi, if i have a task that can take in a very large number of parameters, is it possible to pass them all in as a collection to the task? eg:

Copy code

with Flow(...) as f:
   param_1 = ...
   param_2 = ...
   ...
   param_n = ...
   x = my_task([param_1, param_2, ..., param_n])

Verun Rahimtoola

02/10/2021, 7:42 PM

and then have prefect detect those params as upstream of the task?

Vincent

02/10/2021, 7:51 PM

Hi - I was wondering if anyone has a workaround this issue? https://github.com/PrefectHQ/prefect/issues/4085

Jonathan Lowery

02/10/2021, 8:16 PM

Is there a (locking?) mechanism to make mapped tasks mutually exclusive at runtime? Use case is pulling data for an ETL from multiple sources and wanting to limit one task at a time attempting to insert to a target table. When the tasks run over each other trying to insert to the same table, it causes contention Architecture right now is: gather sources to poll for updates -> map(poll for updates) -> map(insert to db) I suppose one alternate approach would be feeding all the gathered tasks through a grouping task that sets them as upstream/downstream from one another based on target table?

James Phoenix

02/10/2021, 9:33 PM

Hey everyone!

James Phoenix

02/10/2021, 9:33 PM

I'm trying to use a prefect.client() to contact my prefect cloud in flask.

James Phoenix

02/10/2021, 9:34 PM

Copy code

from clean_energy.client_wrappers.prefect_flows.flows import test #THIS IS A FLOW
from flask import (Blueprint)

prefect_client_routes = Blueprint('prefect_client_routes', __name__, url_prefix="/api/prefect")

import prefect
from config.settings import Config

client = prefect.Client(api_token=Config.PREFECT_USER_TOKEN)

@prefect_client_routes.route('/')
def testing():
    return test.register(project_name='clean_energy')

James Phoenix

02/10/2021, 9:34 PM

Currently i'm getting:

James Phoenix

02/10/2021, 9:34 PM

Copy code

prefect.utilities.exceptions.ClientError: [{'path': ['project'], 'message': 'field "project" not found in type: \'query_root\'', 'extensions': {'path': '$.selectionSet.project', 'code': 'validation-failed', 'exception': {'message': 'field "project" not found in type: \'query_root\''

James Phoenix

02/10/2021, 9:34 PM

I know:

James Phoenix

02/10/2021, 9:34 PM

• The project exists on Prefect Cloud

James Phoenix

02/10/2021, 9:35 PM

• My API_token is a USER token

James Phoenix

02/10/2021, 9:35 PM

What am I doing wrong?

Shawn Marhanka

02/10/2021, 10:18 PM

Hi Prefect! Is there a way to create new read only (or restricted) users in Prefect Cloud? My account is saying I have reached the limit and cannot add anyone else. I have a team member that just needs to view the flow schematic, but doesn’t need access to perform any actions. Thank you

Danny Vilela

02/11/2021, 3:27 AM

Hi Prefect! Quick question about flow parameters: if I wanted to know what parameters a flow “accepts”, I’m aware I can do

flow.parameters() -> Set[Parameter]

. But it seems like the implementation is “shallow,” in that it just checks for top-level tasks that are parameters (as opposed to recursively pulling parameters from tasks):

Copy code

def parameters(self) -> Set[Parameter]:
    """
    Returns any parameters of the flow.

    Returns:
        - set: a set of any Parameters in this flow
    """
    return {p for p in self.tasks if isinstance(p, Parameter)}

Is there a reasonable way to extend the Flow class to search for all parameters declared within the

Flow(..) as flow

context manager?

sark

02/11/2021, 7:02 AM

hello everyone, i was wondering if for a CI setup if it is good, to not only make it register flows to ensure that the deployment reflects what is in code that way, but also when a flow is removed from the repo that it gets automatically deleted at prefect as well? or does this interact badly with how prefect is designed and the intended development workflow?

Karolína Bzdušek

02/11/2021, 8:44 AM

Not sure where to post this, few of my notes regarding docs • at page about Secrets, one want to click to redirect to the link about AWS/GitHub/Twitter Credentials, and got only 404 error. (https://docs.prefect.io/orchestration/concepts/secrets.html) • As I am trying to find out what GitHub token I should store (GitHub is a storage under company account in private repo), I found out that one of the option of the storage is "ref", which I somewhere (I forgot where in docs) found out is set to default "master". I think it would be nice to have this mentioned, the was one of the reasons why it couldn't find the repo (our default name for the branch is "main", doesn't occur to me that could be a problem).

Laura Vaida

02/11/2021, 9:39 AM

hi folks, what ist the actual equivalent to this?: flow.environment = DaskKubernetesEnvironment()

Jan Marais

02/11/2021, 9:48 AM

I see in the docs that the default message MIME type for the

EmailTask

is text/html but it seems that this is hard coded. Would it make sense to have it as a parameter to the task in the future? I just have some plain text to be sent and gmail is not parsing my newlines properly.

Ruben

02/11/2021, 11:22 AM

Hi there, I was send here by support. How can I register a flow to prefect Core and then let it run once? I don’t need any scheduled reruns or anything. Just a single run on an agent polling for this flow. I know it can be done using the UI, but I’d rather do this from python.

Gareth Dwyer

02/11/2021, 11:56 AM

Hey all, noob question - I’m trying to run through the hello world tutorial but using Docker on a VPS instead of locally and I get

Copy code

404 Client Error for <http+docker://localhost/v1.40/containers/create>: Not Found ("No such image: prefecthq/prefect:0.14.7")

Did I do something wrong or is this related to the new 14.7 release that I see on GitHub?

Adam

02/11/2021, 12:42 PM

Hi @Jim Crist-Harif and team, read some of your earlier replies, but wanted to ask whether by any chance you have had time to publish a best practice blog post or repo that shows an opinionated way of having a project that: • Uses a base docker image that contains prefect, all the custom deps and shared code • Stores the flows on Github / s3 / etc • Can organise individual flows into groups/folders locally to keep things a bit clean (and play well with pythonpath etc) and ideally have some convention in individual flow files to indicate which project they should be registered with. • Some type of CI/CD process that does the build of the image (when needed) and stores the flows • Uses Kubernetes (with or without Dask) @Chris White @Jeremiah I’ve been using Prefect for about 6 months now and I have many of the above aspects done but it’s all becoming quite a big mess. We’re looking for an opinionated / best practice way to do it now. Based on a lot of the questions above, I think a cookie cutter with sane defaults would really benefit the community. WDYT? P.S. Would be more than happy to show you the mess we have and get your advice on how things can be improved

➕ 2

🎯 3

Nikul

02/11/2021, 2:16 PM

Hi All, I'm trying to run a flow using LocalDaskExecutor (using LocalDaskExecutor()), but I get an error: 'signal only works in main thread'. I was wondering if anyone knows what the problem could be. I get similar issues with DaskExecutor also. I'm running prefect==0.14.5, dask==2021.1.1, distributed==2020.12.0 Thank you

👀 1

Michael Hadorn

02/11/2021, 4:26 PM

I use a docker run with gitlab storage. I got this error then:

Copy code

prefect.utilities.exceptions.ClientError: 400 Client Error: Bad Request for url: <http://host.docker.internal:4200/graphql>

The following error messages were provided by the GraphQL server:

    INTERNAL_SERVER_ERROR: Variable "$input" got invalid value null at
        "input.states[0].task_run_id"; Expected non-nullable type UUID! not to be null.

The GraphQL query was:

    mutation($input: set_task_run_states_input!) {
            set_task_run_states(input: $input) {
                states {
                    status
                    message
                    id
            }
        }
    }

The passed variables were:

    {"input": {"states": [{"state": {"message": "Starting task run.", "cached_inputs": {}, "context": {"tags": []}, "_result": {"__version__": "0.14.6", "type": "NoResultType"}, "__version__": "0.14.6", "type": "Running"}, "task_run_id":
 null, "version": null}]}}

locally and also with the local agent on the ui server it's working.

Asif Imran

02/11/2021, 5:27 PM

Does

<http://cloud.prefect.io|cloud.prefect.io>

offer two-factor authentication?

Luis Gallegos

02/11/2021, 5:34 PM

Hi all, i want to know if this is possible. Run via dask executor different projects each one with respective virtual environment, in order to encapsulate its dependencies version. Thanks for ur help. A little representation

Milly gupta

02/11/2021, 10:27 PM

Hi, does anyone know if there are any free read only users available on the Prefect Cloud Team Plan?

Marwan Sarieddine

02/11/2021, 11:58 PM

Hi folks, I am trying to construct a graphql query to aggregate the total number of successful task runs that we had in the past month but I am noticing that it is not so trivial to form the query so I was wondering if someone else might have already done so and is willing to share the query they used. I imagine given the intended update in prefect’s pricing that others will also want to run similar queries (in case these analytics aren’t shown somewhere on the prefect cloud UI) to track their usage.