read_sql.py

#!/usr/bin/env python

from typing import Callable, Any
import warnings

import funcy as fy
import pandas as pd
import pymysql


warnings.filterwarnings(
    'ignore', message='.*other DBAPI2 objects are not tested.*'
)

Connection = Any  # Hint by variable.
SQL = str


@fy.memoize
def _sql_reader(connect: Callable[[], Connection]) -> Callable[[SQL], pd.DataFrame]:
    return fy.partial(pd.read_sql, con=connect())


def sql_reader(connect: Callable[[], Connection]) -> Callable[[SQL], pd.DataFrame]:
    def read_sql(*args, **kwargs):
        return _sql_reader(connect)(*args, **kwargs)

    return read_sql


def mysql_connector(port: int, db: str) -> Callable[[], Connection]:
    return lambda: pymysql.connect(
        host='localhost',
        port=port,
        db=db,
        ...
    )


# $ ssh DEST -L PORT:HOST:PORT -Nf  # DEST-HOST-mysql1
read_sql_mysql_1 = sql_reader(mysql_connector(PORT, DB))

# $ ssh DEST -L PORT:HOST:PORT -Nf  # DEST-HOST-mysql2
read_sql_mysql_2 = sql_reader(mysql_connector(PORT, DB))

# $ ssh DEST -L PORT:HOST:PORT -Nf  # DEST-HOST-BIGDATA
read_sql_bigdata = sql_reader(lambda: BIGDATA.connect(...))