I'm trying to use a context manager to open an SSH tunnel forwarder when Metaflow FlowSpec subclass instances are run locally, but somehow instantiating a FlowSpec instance in the context manager calls the context manager twice causing the instantiation of the FlowSpec class to fail because the ssh tunnel is already in use. How can I prevent Metaflow from causing the context manager to get called twice? Running the following flow fails because two attempts to open the ssh tunnel are made despite the context manager only getting called once. ## Failing Example

from metaflow import FlowSpec
from metaflow import step

from my_pkg.library.databases import initialize_database
from my_pkg.library.ssh import SshTunnel
from my_pkg.settings.databases import DatabaseConfiguration


class TunnelingPipelineExample(FlowSpec):

    @step
    def start(self):
        self.next(self.query_tables)

    @step
    def query_tables(self):
        db, cursor = initialize_database()
        cursor.execute("USE some_database;")
        cursor.execute("SELECT * FROM some_table;")
        for result in cursor.fetchall():
            print(f"RESULT: {result}")
        self.next(self.end)

    @step
    def end(self):
        print(f"Completed pipeline!")


if __name__ == "__main__":
    with SshTunnel(DatabaseConfiguration()):
        TunnelingPipelineExample()

## Working Example

from my_pkg.library.databases import initialize_database
from my_pkg.library.ssh import SshTunnel
from my_pkg.settings.databases import ApplicationDatabaseUsa

if __name__ == "__main__":
    with SshTunnel(DatabaseConfiguration()):
        db, cursor = initialize_database()
        cursor.execute("USE some_database;")
        cursor.execute("SELECT * FROM some_table;")
        for result in cursor.fetchall():
            print(f"RESULT: {result}")

The same code succeeds without Metaflow, so I assume Metaflow is somehow causing the context manager to get called twice given the error message says the tunnel is already in use.

metaflow calls itself so the

with

statement would run multiple times. You could possibly do something like:

if sys.argv[2] == "run" # Or whatever the proper number
  with SshTunnel(...)
    TunnelingPipelineExample()
else:
  TunnelingPipelineExample()

the trick will be introspecting the command line properly to make it work. It should only call “run” once (when you do it). then it calls things like “step” or “batch”.

@dry-beach-38304, thank you for your feedback. It helps that you confirm Metaflow is causing the context manager to get called twice. How does introspecting to catch the run command help with the second call?

The second time, it will call itself with a different command, something like

myflow.py step …

instead of

myflow.py run

so it should only call itself once with

run

.

Ah, thank you!!! That makes sense now, I'll try that out.