Cursors Port Forwarder

Abstract base class for port forwarding implementations.

Overview

Port forwarders are responsible for establishing tunnels to remote databases through various protocols (e.g., AWS SSM, SSH). This allows Cursor objects to connect to databases that are not directly accessible from the local machine.

To create a custom port forwarder, you must subclass PortForwarder and implement the setup and teardown methods.

Example

Here is a conceptual example of a custom port forwarder:

from clearskies.cursors.port_forwarding import PortForwarder


class MyCustomForwarder(PortForwarder):
    def setup(self, original_host: str, original_port: int) -> tuple[str, int]:
        print(f"Setting up tunnel to {original_host}:{original_port}...")
        # In a real implementation, you would start a subprocess or thread
        # to establish the tunnel here.
        local_port = 12345
        print(f"Tunnel established on localhost:{local_port}")
        return ("localhost", local_port)

    def teardown(self) -> None:
        print("Tearing down tunnel...")
        # Clean up any resources (e.g., terminate subprocess).


# This forwarder can then be passed to a cursor configuration:
forwarder = MyCustomForwarder()
# mysql_cursor = clearskies.cursors.Mysql(
#     hostname="private-db.internal",
#     port=3306,
#     port_forwarding=forwarder,
# )

Table of contents