db

detdevlib.etl.db

DialectSettings

Bases: ABC

Abstract base class for dialect-specific settings.

get_connection_url abstractmethod

get_connection_url() -> str

Constructs and returns a SQLAlchemy connection URL.

Returns:

Name	Type	Description
str	str	A SQLAlchemy connection URL.

get_procedures_sql

get_procedures_sql(
    schema: Optional[str] = None,
) -> Optional[tuple[str, dict]]

Returns the SQL query to list stored procedures.

Parameters:

Name	Type	Description	Default
schema	Optional[str]	The schema to query.	None

Returns:

Type	Description
Optional[tuple[str, dict]]	Optional[tuple[str, dict]]: A tuple with the SQL query string and a dictionary of parameters, or None if not supported.

get_engine_kwargs

get_engine_kwargs() -> dict

Returns arguments to be passed to sqlalchemy.create_engine.

Useful for dialect-specific optimizations like fast_executemany.

get_insert_kwargs

get_insert_kwargs() -> dict

Returns dialect-specific parameters for pd.to_sql.

Returns:

Name	Type	Description
dict	dict	Parameters to override the defaults in DatabaseManager. Example: {'method': None} or {'chunksize': 1000}

MSSQLSettings

Bases: BaseSettings, DialectSettings

Configuration for 'mssql' dialect.

get_engine_kwargs

get_engine_kwargs() -> dict

Enables the specialized high-performance insert for MSSQL.

PostgreSQLSettings

Bases: BaseSettings, DialectSettings

Configuration for the 'postgresql' dialect.

get_insert_kwargs

get_insert_kwargs() -> dict

Override method to use COPY instead of multi-row INSERT.

SQLiteSettings

Bases: BaseSettings, DialectSettings

Configuration for the 'sqlite' dialect.

MySQLSettings

Bases: BaseSettings, DialectSettings

Configuration for the 'mysql' dialect.

SnowflakeSettings

Bases: BaseSettings, DialectSettings

Configuration for the 'snowflake' dialect.

DatabaseManager

A database manager that is agnostic to the underlying dialect.

engine property

engine: Engine

Provides direct access to the underlying SQLAlchemy Engine.

Returns:

Name	Type	Description
Engine	Engine	The active SQLAlchemy engine instance.

Raises:

Type	Description
RuntimeError	If engine is not active.

__init__

__init__(settings: DialectSettings)

Initializes the DatabaseManager.

Parameters:

Name	Type	Description	Default
settings	DialectSettings	The dialect-specific settings.	required

connect

connect() -> Self

Initializes the SQLAlchemy Engine.

Returns:

Type	Description
Self	The instance of the DatabaseManager.

disconnect

disconnect()

Disposes of the SQLAlchemy engine.

execute_statement

execute_statement(
    sql: str, params: Optional[dict] = None
) -> None

Executes an arbitrary SQL statement.

This method is suitable for operations like CREATE, INSERT, UPDATE, DELETE that do not return a result set.

Parameters:

Name	Type	Description	Default
sql	str	The SQL statement to execute.	required
params	Optional[dict]	Parameters for the SQL statement.	None

execute_query

execute_query(
    sql: str, params: Optional[dict] = None
) -> pd.DataFrame

Executes a SQL query and returns results as a DataFrame.

Parameters:

Name	Type	Description	Default
sql	str	The SQL query to execute.	required
params	Optional[dict]	Parameters for the SQL query.	None

Returns:

Type	Description
DataFrame	A pandas DataFrame with the query results.

list_tables

list_tables(schema: Optional[str] = None) -> list[str]

Retrieves a list of table names for a given schema.

describe_table

describe_table(
    table_name: str, schema: Optional[str] = None
) -> pd.DataFrame

Retrieves column metadata (name, type, etc.) for a specific table.

insert_df

insert_df(
    df: DataFrame,
    table: str,
    schema: Optional[str] = None,
    if_exists: Literal["fail", "append"] = "fail",
) -> None

Performs a fast, bulk insert of a pandas DataFrame into a database table.

list_stored_procedures

list_stored_procedures(
    schema: Optional[str] = None,
) -> list[str]

Retrieves a list of stored procedure names for a given schema.

Note: This is often a dialect-specific operation.