Sharing non-serializable objects across tasks within a Flow

[Mastermjr]

Issue from discussion with @cicdw on slack

I'm using arangodb for storing different values of objects/actions within my tasks. In order to make my life easier, I want to create one instance to a DB wrapper to be shared across all tasks within a given flow like below:

@task()
def connection():
    prefect.context['db'] = DATABASE(url, user, pass)
@task()
def operation():
    db = prefect.context['db']
    # do stuf.... 
    db.send('{"asdf": aaaa}')
with Flow('stuff') as flow: 
    connection()
    operation()
    close_connection()

This is not very fleshed out, but chris asked I start a discussion here. The reason I desire this is two fold:

1. Efficiency: to my very fledgling knoweledge of python development, it will be more efficient to create one object across all my tasks, instead of creating an object in each task (If I had time it may be more efficient to just do api calls within each task as a function, but time is limited)
2. Code re-use/reading: I think having one database connection to share across my tasks would be really nice, and make the code more resusable/modular, and make it easier to read

I think theres an idea for resource management, which is nice, but I would still have to share a reference to the object as a parameter for all my task functions, which I don't want to do (which maybe I should just do anyway?).

My workaround for now will look like this, which is similar to above, but I like the modularity/code resuse that the above allows.

@task()
def connection():
    db = DATABASE(url, user, pass)
    return db
@task()
def operation(db):
    # do stuf.... 
    db.send('{"asdf": aaaa}')

@task()
def operation(db):
    db.close()

with Flow('stuff') as flow: 
    db = connection()
    operation(db)
    close_connection(db)

[cicdw]

Hi @Mastermjr - you should check out the proposal in #2891 (comment)