https://prefect.io logo
Powered by
Title
s

Samuel Hinton

03/08/2023, 10:36 AM
Ah, I think I’ve found another bug, this one in flow execution instead of the UI. I believe prefect does not work with function decorators when using the 
@wraps(fn)
annotation. Full reproduction of the issue below, but TLDR if you keep the @wraps you get an exception of 
TypeError: add() takes 2 positional arguments but 3 were given
. If you remove the @wraps, the flow runs as expected. Happy to raise this on GitHub, just again trying to validate here before adding more issues to triage 
from prefect import flow, get_run_logger, task
from functools import wraps

def decorator(fn):
    @wraps(fn)  # With wraps, the flow does not work. Comment this line out and it will work again.
    def wrapper(a, b, **kwargs):
        return fn(a, b, **kwargs)
    return wrapper

@task
@decorator
def add(a, b, logger=None):
    result = a + b
    if logger is not None:
        <http://logger.info|logger.info>(f"Adding {a} and {b} to get {result}")
    return result

@flow
def kwarg_flow():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Starting an ECS task")
    a, b = 1, 2
    add.submit(a, b, logger=logger)
    <http://logger.info|logger.info>("Everything is done!")


if __name__ == "__main__":
    from prefect.deployments import Deployment
    from prefect.filesystems import S3
    from prefect.settings import PREFECT_API_URL, temporary_settings

    with temporary_settings({PREFECT_API_URL: "<http://orion.orchestrator-ds-dat3.arenko:4200/api>"}):
        s3_block = S3.load("flow-storage")
        print(f"Deploying flow")
        d = Deployment.build_from_flow(
            flow=kwarg_flow,
            name="default",
            storage=s3_block,
            work_pool_name="dask",
            load_existing=False,
            skip_upload=False,
            apply=True,
            path="/",
        )
n

Nate

03/08/2023, 4:33 PM
Hi @Samuel Hinton I think you want 
def decorator(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        return fn(*args, **kwargs)
    return wrapper
and with that if i run the flow I get 
10:31:56.636 | INFO    | prefect.engine - Created flow run 'charcoal-numbat' for flow 'kwarg-flow'
10:31:57.235 | INFO    | Flow run 'charcoal-numbat' - Starting an ECS task
10:31:57.239 | INFO    | Flow run 'charcoal-numbat' - Everything is done!
10:31:57.381 | INFO    | Flow run 'charcoal-numbat' - Created task run 'add-0' for task 'add'
10:31:57.383 | INFO    | Flow run 'charcoal-numbat' - Submitted task run 'add-0' for execution.
10:31:57.698 | INFO    | Flow run 'charcoal-numbat' - Adding 1 and 2 to get 3
10:31:57.817 | INFO    | Task run 'add-0' - Finished in state Completed()
10:31:57.964 | INFO    | Flow run 'charcoal-numbat' - Finished in state Completed('All states completed.')
note you could also decorate async functions with the same decorator if you did 
from prefect.utilities.asyncutils import is_async_fn

def decorator(fn):
   if is_async_fn(fn):
       @wraps(fn)
       async def wrapper(*args, **kwargs):
           # do interesting things
           return await fn(*args, **kwargs)
       return wrapper
   else:
       @wraps(fn)
       def wrapper(*args, **kwargs):
           # do interesting things
           return fn(*args, **kwargs)
       return wrapper
s

Samuel Hinton

03/09/2023, 2:06 AM
Hmm, yeah, changing my code to: 
def wrapper(a, b, *args, **kwargs)
Did fix the issue (I do need a and b in the wrapper, even though I dont in the toy example, and better to pull them out explicitly instead of args[0], etc). However, both decorators are completely valid python, so its odd prefect crashes with one but works with the other. Do you have insight as to why this is (Id be happy to open a PR, Im just not even sure how to get started_
n

Nate

03/09/2023, 2:09 AM
prefect crashes with one but works with the other
hmm, do you mind giving an MRE for each of these here so I know for sure what you mean?
s

Samuel Hinton

03/09/2023, 2:09 AM
Sure thing, one tick
Python file attached with reproduction. Summary: When running locally, kwargs are passed in as kwargs, and both decorators work. But when running the prefect flow, either locally or in the cloud, prefect incorrectly turns my defined kwargs and passes them in as args> Unfortunately this means that there are plenty of ways for decorators to break (actually using the *args, changing the order of kwargs, etc, all will break it)
example_kwarg_crashes.py
@Nate again happy to try and look into this and make a PR provided a) Im correct that theres an issue and b) someone can point me in the right direction .The prefect repo is big
n

Nate

03/09/2023, 2:45 AM
hmm it looks like the first file (that didnt crash i assume) is an accessible, you can just paste it in here if you want
s

Samuel Hinton

03/09/2023, 2:46 AM
Oh sorry I had two files, but to make it easy I turned the “2 flows, 1 task each” into one flow with two tasks, one which crashes and the other which works
👍 1
So that file should be all you need. Can get rid of the deployment stuff at the bottom too, Ive verified it plays up when run locally
@Nate any luck reproducing it on your machine?
n

Nate

03/09/2023, 3:44 AM
yeah 
2023-03-08 21:39:19.387 | ERROR    | __main__:<module>:63 - add_noargs() takes 2 positional arguments but 3 were given
tried / failed to logic my way through it quickly, will be a tomorrow thing for me though as its almost 10pm CST 😄 - will come back to it in the morning
s

Samuel Hinton

03/09/2023, 3:50 AM
Awesome, if I can figure out the logic today too Ill make a GH issue and link it here to try and grease your wheels
Alright, making progress. Reproduction is now down to only a few lines; 
from prefect.utilities.callables import parameters_to_args_kwargs, get_call_parameters


def fn(a, b, c=3):
    return a + b + c


parameters = get_call_parameters(fn, (1, 2), {"c": 100})
print(parameters)
args, kwargs = parameters_to_args_kwargs(fn, parameters)
print(args, kwargs)
Figured out the issue. https://github.com/PrefectHQ/prefect/issues/8753
n

Nate

03/09/2023, 4:56 PM
nice digging!
On that note, this feels more like an issue with Python itself than prefect, and I've actually raised with CPython as well here as well.
i had a similar thought when reading your summary - thanks again for making the issue!
#prefect-communityJoin Slack
Powered by