Hey, not sure if this the right place to ask, can anyone provide an idiomatic example of using Oracle DB connection across multiple tasks please? I am helping out a colleague with an ETL flow which requires multiple queries done across multiple connections, which would preferably persist throughout the flow. The initial problem we are encountering is that you cannot pass the connection object between tasks as it needs to be serialisable.

I have an example for Snowflake, but you can use the same for Oracle:

from prefect import resource_manager

@resource_manager
class SnowflakeConnection:
    def __init__(self, database: str = "DEV"):
        self.database = database

    def setup(self):
        db_conn_string = get_snowflake_connection_string(self.database)
        db_engine = create_engine(db_conn_string)
        return db_engine.connect()

    @staticmethod
    def cleanup(conn):
        conn.close()

woo, thanks so much for a speedy reply! ❤️

Did not know about this feature so I'm taking the liberty of asking questions about it : how does it work with parallel tasks ? For example if I'm running mapped tasks on a Dask cluster. Is there a way to deal with potential concurrency ?

good question - afaik a database connection cannot be cloud-pickled and shared between Dask workers. Let me ask the team to be sure

Thanks !

@Sylvain Hazard it looks like you’re right - this wouldn’t work with Dask, you would have to define the connection within the task.

Noted, thanks for looking out !

@Anna Geller we have tried using the above example as

import prefect
import cx_Oracle
from prefect import task
from prefect import resource_manager


@resource_manager
class ConnectionFromSecret:
    def __init__(self, secret):
        self.host = secret['host']
        self.port = secret['port']
        self.dbname = secret['dbname']
        self.username = secret['username']
        self.password = secret['password']

    def setup(self):
        dsn = cx_Oracle.makedsn(self.host, self.port, self.dbname)
        return cx_Oracle.connect(user=self.username, password=self.password, dsn=dsn)

    @staticmethod
    def cleanup(conn):
        conn.close()

and

with ConnectionFromSecret(secret) as conn:
  ...

which works fine locally with

flow.run

but fails on the server with

Unexpected error: TypeError('Object of type ConnectionFromSecret is not JSON serializable')
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 926, in get_task_run_state
    result = self.result.write(value, **formatting_kwargs)
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/results/prefect_result.py", line 62, in write
    new.location = self.serializer.serialize(new.value).decode("utf-8")
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/serializers.py", line 110, in serialize
    return json.dumps(value).encode()
  File "/usr/local/lib/python3.7/json/__init__.py", line 231, in dumps
    return _default_encoder.encode(obj)
  File "/usr/local/lib/python3.7/json/encoder.py", line 199, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "/usr/local/lib/python3.7/json/encoder.py", line 257, in iterencode
    return _iterencode(o, 0)
  File "/usr/local/lib/python3.7/json/encoder.py", line 179, in default
    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type ConnectionFromSecret is not JSON serializable

can you show your Flow object definition?

might need a few mins to tidy the flow to paste it here

I see - you use result=PrefectResult() - this is the problem here

which result should be used?

can you try using this result class only for the tasks that absolutely require storing their results in a Prefect backend? since you assigned that on a flow level, Prefect tries to serialize results of all tasks using PrefectResult. However, this result class can only be used for JSON serializable return values from tasks and the connection from the resource manager is not JSON serializable