| 116 | |
| 117 | |
| 118 | class PostgresDataLoader(DataLoader): |
| 119 | def __init__( |
| 120 | self, |
| 121 | db_name: str, |
| 122 | db_host: Optional[str] = None, |
| 123 | db_port: Optional[int] = None, |
| 124 | db_username: Optional[str] = None, |
| 125 | db_password: Optional[str] = None, |
| 126 | ): |
| 127 | """ |
| 128 | The main data reading object. |
| 129 | |
| 130 | Parameters |
| 131 | ---------- |
| 132 | db_name : str |
| 133 | The database name (or path if sqlite is used) where the given tables are stored. |
| 134 | db_host : str |
| 135 | The IP address/host name of the database server. |
| 136 | db_port : str |
| 137 | The port number of the database server. |
| 138 | db_username : str |
| 139 | The username used to connect to the database server. |
| 140 | db_password : str |
| 141 | The password used to connect to the database server. |
| 142 | """ |
| 143 | self.db_type = DBType.POSTGRESQL |
| 144 | |
| 145 | self.__db_host = db_host |
| 146 | self.__db_port = db_port |
| 147 | self.__db_username = db_username |
| 148 | self.__db_password = db_password |
| 149 | self.__db_name = db_name |
| 150 | |
| 151 | def get_db_engine(self) -> Optional[sqlalchemy.engine.Engine]: |
| 152 | """ |
| 153 | This method creates a new SQLAlchemy engine useful to create database connections. |
| 154 | |
| 155 | Returns |
| 156 | ------- |
| 157 | sqlalchemy.engine.Engine |
| 158 | A new SQLAlchemy connection engine. |
| 159 | Returns None if the database type is unrecognised. |
| 160 | |
| 161 | """ |
| 162 | return sqlalchemy.create_engine( |
| 163 | "postgresql://{}:{}@{}:{}/{}".format( |
| 164 | self.__db_username, |
| 165 | self.__db_password, |
| 166 | self.__db_host, |
| 167 | self.__db_port, |
| 168 | self.__db_name, |
| 169 | ) |
| 170 | ) |
| 171 | |
| 172 | def get_counts( |
| 173 | self, table_name: str, schema_name: Optional[str] = None |
| 174 | ) -> Dict[str, Tuple[int, int]]: |
| 175 | """ |
nothing calls this directly
no outgoing calls
no test coverage detected