InfDBClient Class

Thin Postgres client backed by psycopg2 with a SQLAlchemy engine helper.

__init__(infdb_config: InfdbConfig, log: logging.Logger, db_name: str = 'postgres') -> None

Initializes the client and opens a connection.

Parameters:

Name	Type	Description	Default
infdb_config	InfdbConfig	Configuration provider for database parameters.	required
log	Logger	Logger instance to use for diagnostics.	required
db_name	str	Database name to connect to (defaults to "postgres").	'postgres'

Raises:

Type	Description
OperationalError	If the connection cannot be established.

__enter__() -> InfdbClient

Enters the context manager (returns self).

__exit__(exc_type, exc, tb) -> None

Exits the context manager and closes resources.

close() -> None

Closes the cursor and connection, ignoring close-time exceptions.

__del__() -> None

Performs best-effort resource cleanup on GC.

__str__() -> str

Returns a human-readable description of the client.

execute_query(query: str, params: ParamsTuple = None) -> List[Row]

Executes SQL and returns fetched rows; returns [] for non-SELECT statements.

Parameters:

Name	Type	Description	Default
query	str	SQL query to execute.	required
params	ParamsTuple	Optional positional parameters for the query.	None

Returns:

Type	Description
List[Row]	A list of rows for SELECT-like statements; empty list for statements without a result set.

execute_sql_files(sql_dir: str, file_list: Optional[Sequence[str]] = None, format_params: Optional[Dict[str, Any]] = None)

Executes a set of .sql files in lexicographic order by default.

Parameters:

Name	Type	Description	Default
sql_dir	str	Directory containing SQL files.	required
file_list	Optional[Sequence[str]]	Optional explicit list of filenames to execute (in given order).	None
format_params	Optional[Dict[str, Any]]	Optional dict for Python str.format substitutions.	None

Raises:

Type	Description
FileNotFoundError	If the directory or a specified file does not exist.
Exception	Re-raises any execution error after rolling back.

execute_sql_file(file_path: str, format_params: Optional[Dict[str, Any]] = None) -> None

Executes a single SQL file by delegating to execute_sql_files.

Parameters:

Name	Type	Description	Default
file_path	str	Absolute or relative path to the SQL file.	required
format_params	Optional[Dict[str, Any]]	Optional dict for Python str.format substitutions.	None

get_db_engine()

Creates and returns a SQLAlchemy engine for the target DB.

Returns:

Type	Description
	A SQLAlchemy Engine connected to the configured database.

get_db_params() -> DBParams

Returns a shallow copy of the resolved DB parameters.

Returns:

Type	Description
DBParams	A dictionary of database parameters.

get_pandas(sql: str, format_params: Optional[Dict[str, Any]] = None) -> Any

Helper method to read SQL query results into a pandas DataFrame.

get_pandas_sqlfile(path: str, format_params: Optional[Dict[str, Any]] = None) -> Any

Helper method to read SQL query files results into a pandas DataFrame.