Is there a centralized way to run code before every deployment or at least every flow? (ex. setup dependency injection, app configuration etc) ps. I had issues wrapping the flow decorator. If anyone has done this successfully, code is welcome. thanks.

hi @Nikolay Tsvetanov - are you using workers? this sounds like roughly the idea behind

pull

steps

I also wanted to run centralized code before deployments/flows and I also took the approach of wrapping the prefect flow decorator. I didn't have any issues wrapping it. Can you share what issues you ran into or some sample code? Our decorator looks like this:

from prefect import flow
from functools import wraps

class Workflow:
  def __init__():
    ...
  
  def flow(self):
    def decorator(func):
      ...
      prefect_wrapper = flow(name=..., ...)
      
      @wraps(func)
      def wrapper(*args, **kwargs):
        # Centralized code here
        return func(*args, **kwargs)

      return prefect_wrapper(wrapper)

    return decorator

Unfortunately I dont have the code anymore. If I remember correctly there were issues with the deployment arguments or data being passed to the flow.

If you're using

async

, then you have to be a bit more careful with this. We have a version of this code that work

async

, but it does a bit more work. Also, I forgot to specify usage:

from workflow_file import Workflow

workflow = Workflow()

@worfklow.flow
def flow(...):
  ...

yeah, we do use async flows

Okay, then your

wrapper

should also be

async