Skip to content

eds_scikit.io.postgres

PostgresData 

PostgresData(dbname: Optional[str] = None, schema: Optional[str] = None, user: Optional[str] = None, host: Optional[str] = None, port: Optional[int] = None)

Bases: BaseData

Source code in eds_scikit/io/postgres.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(
    self,
    dbname: Optional[str] = None,
    schema: Optional[str] = None,
    user: Optional[str] = None,
    host: Optional[str] = None,
    port: Optional[int] = None,
):
    (
        self.host,
        self.port,
        self.dbname,
        self.user,
    ) = self._find_matching_pgpass_params(host, port, dbname, user)
    self.schema = schema

read_sql 

read_sql(sql_query: str, **kwargs) -> pd.DataFrame

Execute pandas.read_sql() on the database.

PARAMETER DESCRIPTION
sql_query

SQL query (postgres flavor)

TYPE: str

**kwargs

additional arguments passed to pandas.read_sql()

DEFAULT: {}

RETURNS DESCRIPTION
df

TYPE: pandas.DataFrame

Source code in eds_scikit/io/postgres.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def read_sql(self, sql_query: str, **kwargs) -> pd.DataFrame:
    """Execute pandas.read_sql() on the database.

    Parameters
    ----------
    sql_query : str
        SQL query (postgres flavor)
    **kwargs
        additional arguments passed to pandas.read_sql()

    Returns
    -------
    df : pandas.DataFrame

    """
    connection_infos = {
        param: getattr(self, param) for param in ["host", "port", "dbname", "user"]
    }
    connection_infos["password"] = pgpasslib.getpass(**connection_infos)
    connection = pg.connect(**connection_infos)
    if self.schema:
        connection.cursor().execute(f"SET SCHEMA '{self.schema}'")

    df = pd.read_sql(sql_query, con=connection, **kwargs)

    connection.close()
    return df
Back to top