Last active
April 1, 2022 01:24
-
-
Save moskytw/9bf0692f4628897431736ea0bb23d6e4 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/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(...)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment