carlogtt_python_library.database.database_sql module

This module …

class carlogtt_python_library.database.database_sql.Database

Bases: ABC, Generic[ConnT]

abstractmethod close_db_connection() → None

abstract property db_connection: ConnT

db_utils: DatabaseUtils

abstractmethod fetch_from_db(sql_query: str, sql_values: Sequence[bool | bytes | float | int | str | Decimal | date | time | datetime | timedelta | struct_time | None] = (), *, fetch_one: bool = False) → Generator[dict[str, Any], None, None]

abstractmethod open_db_connection() → None

abstractmethod send_many_to_db(sql_query: str, sql_values: Iterable[Sequence[bool | bytes | float | int | str | Decimal | date | time | datetime | timedelta | struct_time | None]]) → None

abstractmethod send_to_db(sql_query: str, sql_values: Sequence[bool | bytes | float | int | str | Decimal | date | time | datetime | timedelta | struct_time | None] = ()) → None

class carlogtt_python_library.database.database_sql.MySQL(host: str, user: str, password: str, port: str, database_schema: str)

Bases: Database[MySQLConnectionAbstract | PooledMySQLConnection]

Handles MySQL database connections.

Parameters:

• host – Hostname or IP address of the MySQL server.

• user – Username to authenticate with the MySQL server.

• password – Password to authenticate with the MySQL server.

• port – Port number of the MySQL server.

• database_schema – Name of the database schema to use.

Attributes

db_utils

Instance of DatabaseUtils (helper for reading external SQL files, etc.).

close_db_connection() → None

Close the MySQL db connection. Auto retry up to 4 times on connection error.

Raises:

MySQLError – If the operation fails.

property db_connection: MySQLConnectionAbstract | PooledMySQLConnection

Gets the active db connection. If there is not an active connection it creates one.

fetch_from_db(sql_query: str, sql_values: Sequence[bool | bytes | float | int | str | Decimal | date | time | datetime | timedelta | struct_time | None] = (), *, fetch_one: bool = False) → Generator[dict[str, Any], None, None]

Fetch data from MySQL database.

Parameters:

• sql_query – SQL query to be executed.

• sql_values – Values to be substituted in the SQL query.

• fetch_one – If True, only fetch the first row.

Returns:

Generator of dictionaries containing the fetched rows.

Raises:

MySQLError – If the operation fails.

open_db_connection() → None

Open a MySQL db connection. Auto retry up to 4 times on connection error.

Raises:

MySQLError – If the operation fails.

send_many_to_db(sql_query: str, sql_values: Iterable[Sequence[bool | bytes | float | int | str | Decimal | date | time | datetime | timedelta | struct_time | None]]) → None

Execute the same SQL statement many times in a single ACID‑compliant transaction. Commit only if every execution succeeds, otherwise roll back.

Parameters:

• sql_query – The parametrized SQL string.

• sql_values – Any iterable yielding values to be substituted in the SQL query.

Raises:

MySQLError – (after rollback) If the operation fails.

send_to_db(sql_query: str, sql_values: Sequence[bool | bytes | float | int | str | Decimal | date | time | datetime | timedelta | struct_time | None] = ()) → None

Send data to MySQL database.

Parameters:

• sql_query – SQL query to be executed.

• sql_values – Values to be substituted in the SQL query.

Raises:

MySQLError – If the operation fails.

class carlogtt_python_library.database.database_sql.PostgreSQL(host: str, user: str, password: str, port: str, database_schema: str)

Bases: Database[None]

Handles PostgreSQL database connections.

Parameters:

• host – Hostname or IP address of the Postgres server.

• user – Username to authenticate with the Postgres server.

• password – Password to authenticate with the Postgres server.

• port – Port number of the Postgres server.

• database_schema – Name of the database schema to use.

Attributes

db_utils

Instance of DatabaseUtils (helper for reading external SQL files, etc.).

close_db_connection() → None

Close the PostgreSQL db connection. Auto retry up to 4 times on connection error.

Raises:

PostgresError – If the operation fails.

property db_connection: None

Gets the active db connection. If there is not an active connection it creates one.

fetch_from_db(sql_query: str, sql_values: Sequence[bool | bytes | float | int | str | Decimal | date | time | datetime | timedelta | struct_time | None] = (), *, fetch_one: bool = False) → Generator[dict[str, Any], None, None]

Fetch data from PostgreSQL database.

Parameters:

• sql_query – SQL query to be executed.

• sql_values – Values to be substituted in the SQL query.

• fetch_one – If True, only fetch the first row.

Returns:

Generator of dictionaries containing the fetched rows.

Raises:

PostgresError – If the operation fails.

open_db_connection() → None

Open a PostgreSQL db connection. Auto retry up to 4 times on connection error.

Raises:

PostgresError – If the operation fails.

send_many_to_db(sql_query: str, sql_values: Iterable[Sequence[bool | bytes | float | int | str | Decimal | date | time | datetime | timedelta | struct_time | None]]) → None

Execute the same SQL statement many times in a single ACID‑compliant transaction. Commit only if every execution succeeds, otherwise roll back.

Parameters:

• sql_query – The parametrized SQL string.

• sql_values – Any iterable yielding values to be substituted in the SQL query.

Raises:

PostgresError – (after rollback) If the operation fails.

send_to_db(sql_query: str, sql_values: Sequence[bool | bytes | float | int | str | Decimal | date | time | datetime | timedelta | struct_time | None] = ()) → None

Send data to PostgreSQL database.

Parameters:

• sql_query – SQL query to be executed.

• sql_values – Values to be substituted in the SQL query.

Raises:

PostgresError – If the operation fails.

class carlogtt_python_library.database.database_sql.SQLite(sqlite_db_path: str | Path, filename: str)

Bases: Database[Connection]

Handles SQLite database connections.

Parameters:

• sqlite_db_path – Fullpath to the SQLite database file.

• filename – Name of the SQLite database file.

Attributes

db_utils

Instance of DatabaseUtils (helper for reading external SQL files, etc.).

close_db_connection() → None

Close the SQLite db connection.

Raises:

SQLiteError – If the operation fails.

property db_connection: Connection

Gets the active db connection. If there is not an active connection it creates one.

fetch_from_db(sql_query: str, sql_values: Sequence[bool | bytes | float | int | str | Decimal | date | time | datetime | timedelta | struct_time | None] = (), *, fetch_one: bool = False) → Generator[dict[str, Any], None, None]

Fetch data from SQLite database.

Parameters:

• sql_query – SQL query to be executed.

• sql_values – Values to be substituted in the SQL query.

• fetch_one – If True, only fetch the first row.

Returns:

Generator of dictionaries containing the fetched rows.

Raises:

SQLiteError – If the operation fails.

open_db_connection() → None

Open a SQLite db connection and cache it for quick access. To equal the style of MySQL it enables some features by default: - Set the cursor to return dictionary instead of tuples. - Enable foreign key constraint.

Raises:

SQLiteError – If the operation fails.

send_many_to_db(sql_query: str, sql_values: Iterable[Sequence[bool | bytes | float | int | str | Decimal | date | time | datetime | timedelta | struct_time | None]]) → None

Execute the same SQL statement many times in a single ACID‑compliant transaction. Commit only if every execution succeeds, otherwise roll back.

Parameters:

• sql_query – The parametrized SQL string.

• sql_values – Any iterable yielding values to be substituted in the SQL query.

Raises:

SQLiteError – (after rollback) If the operation fails.

send_to_db(sql_query: str, sql_values: Sequence[bool | bytes | float | int | str | Decimal | date | time | datetime | timedelta | struct_time | None] = ()) → None

Send data to SQLite database.

Parameters:

• sql_query – SQL query to be executed.

• sql_values – Values to be substituted in the SQL query.

Raises:

SQLiteError – If the operation fails.