dbpd package

Submodules

dbpd.access module

class dbpd.access.Access(filepath: str, fernet_encryption_key: bytes | None = None, description: str = 'MS Access database connection', show_description: bool = True, **pyodbc_connection_kwargs)[source]

Bases: BaseDBPD

Will connect to (or create) an MS Access database, methods can then be used from BaseDBPD

Parameters:

  • filepath – The filepath of the Access database (or where the new one should be created)

  • description – A description of the database (for reference utility only)

  • show_description – Whether the description should be printed to the console upon init

  • pyodbc_connection_kwargs – Any keyword arguments that should be passed to the pyodbc.connect() function

dbpd.dbpd module

class dbpd.dbpd.BaseDBPD(connection_credentials: str | None = None, description: str | None = None, database_type: str | None = None, postgres_schema: str | None = None, mysql_database_name: str | None = None, filepath: str | None = None, sqlite_in_memory: bool | None = False, show_description: bool = True, fernet_encryption_key: bytes | None = None, **connection_kwargs)[source]

Bases: object

This class is the parent class for six child classes that are specific to different database types:

Access MySQL Oracle Postgres SQLite SQLiteInMemory

Abstractions have been created such that working with these various database types is consistent throughout the user’s code base.

The main theme of this class is establish a working relationship between relational databases and Pandas DataFrames; because these objects are tabular in nature, it reveals itself to be an efficient way to manage and manipulate a given database

The query() method of this class is responsible for returning SELECT sql statements as their respective DataFrames. However the query() method can also be used to make changes to the database (such as INSERT, UPDATE, DELETE, etc.)

Users are expected to write their own sql statements to query the database, and parameterized queries are accepted and encouraged as well.

This is not meant to be an Object-Relational-Mapper (ORM) and has no such functionality, although it may be possible for users to create their own ORM using the classes herein.

See the README for examples of how to get started

exception AccessFileError(filepath: str)[source]

Bases: ValueError

COLUMN_NAME_QUERIES = {'mysql': "\n            SELECT \n                column_name AS name\n            FROM\n                information_schema.columns\n            WHERE \n                table_schema = '{database_name}' \n                AND table_name = '{table_name}'\n        ", 'oracle': "\n            SELECT \n                column_name AS name\n            FROM \n                user_tab_columns \n            WHERE \n                table_name = '{table_name}'\n            ORDER BY\n                column_name\n        ", 'postgres': "\n            SELECT \n                column_name AS name\n            FROM \n                information_schema.columns \n            WHERE \n                table_schema || '.' || table_name = '{table_name}'\n            ORDER BY\n                column_name\n\n        ", 'sqlite': '\n            PRAGMA table_info({table_name})\n        '}
exception EncryptionKeyError[source]

Bases: AttributeError

exception ExcelFileError(filepath: str)[source]

Bases: ValueError

exception InitFileError(filepath: str)[source]

Bases: ValueError

exception SqliteFileError(filepath: str)[source]

Bases: ValueError

TABLE_NAME_QUERIES = {'mysql': "\n            SELECT \n                table_name AS name \n            FROM \n                information_schema.tables\n            WHERE \n                table_schema = '{database_name}'\n        ", 'oracle': '\n            SELECT \n                table_name AS name\n            FROM \n                user_tables \n            ORDER BY \n                table_name\n        ', 'postgres': "\n            SELECT \n                table_schema || '.' || table_name AS name\n            FROM \n                information_schema.tables \n            WHERE\n                table_schema = '{schema}'\n            ORDER BY \n                table_schema, \n                table_name\n        ", 'sqlite': "\n            SELECT \n                name\n            FROM \n                sqlite_schema \n            WHERE \n                type = 'table' \n            ORDER BY \n                name\n        "}
classmethod access_driver(filepath: str) → str[source]

Creates the MS Access Driver string that can be used by pyodbc.connect()

Parameters:

filepath – The filepath of the Access database

Returns:

str - The filepath with backward slashes replaced with forward slashes

close(commit_on_quit: bool = True) → None[source]

Closes the current session with the database.

Parameters:

commit_on_quit – Boolean indicating if a final commit should be transacted before close

Returns:

None

column_names(table_name: str, show_names: bool = False) → list[source]

Returns a list of column names for a given table.

Parameters:

  • table_name – The name of the table in which the column names should be queried

  • show_names – Boolean indicating if the column names should be printed to the console

Returns:

list - The list of the column names

commit() → None[source]

Commits current transactions to the database :return: None

classmethod convert_numpy_value(value: Any, column: str | None = None, null_zeroes_for_columns: list | None = None) → Any[source]

Possibly converts a numpy value (typically this value comes from a DataFrame) to its respective Python value

Parameters:

  • value – The value to be possibly be converted

  • column – The column name of the DataFrame in which the value resides - This is only used for nulling zero values

  • null_zeroes_for_columns – A list of columns where zeros should be nullified

Returns:

The converted value

classmethod create_access_database(filepath: str, create_table_schema_sql_list: list | None = None) → str[source]

Creates a new MS Access database, this method can also accept a list of CREATE TABLE sql statements in which to define the database table schema.

Parameters:

  • filepath – The filepath at which the database should be created

  • create_table_schema_sql_list – The list of CREATE TABLE sql statements

Returns:

The filepath with forward slashes replaced with backslashes

classmethod create_sqlite_database(filepath: str, create_table_schema_sql_list: list | None = None) → str[source]

Creates a new SQLite database, this method can also accept a list of CREATE TABLE sql statements in which to define the database table schema.

Parameters:

  • filepath – The filepath at which the database should be created

  • create_table_schema_sql_list – The list of CREATE TABLE sql statements

Returns:

The absolute path of the filepath

decrypt_value(value: bytes, original_encoding: str = 'utf-8', ttl: int | None = None) → Any[source]

Users can pass a Fernet Encryption Key to the constructor of this class, possibly in the case that fields within the database are encrypted. Using this key, this method will decrypt a value. If no key is passed in the constructor (or not set after the fact) this method will raise an Exception.

Parameters:

  • value – The value to be decrypted

  • original_encoding – The original encoding of the value

  • ttl – The timeout of the key

Returns:

str - The string representation of the decrypted value

static decrypt_value_static(fernet_encryption_key: bytes, value: bytes, original_encoding: str = 'utf-8', ttl: int | None = None) → str[source]

This static method can be used directly if the user has an encryption key but would not like to start a database session

Parameters:

  • fernet_encryption_key – The encryption key

  • value – The value to be decrypted

  • original_encoding – The original encoding of the value

  • ttl – The timeout of the key

Returns:

str - The string representation of the decrypted value

drop_all_tables(commit: bool = True) → None[source]

Drops all tables from the database.

Parameters:

commit – Boolean indicating if this dropping transaction should be committed

Returns:

None

drop_table(table_name: str, commit: bool = True)[source]

Drops a table from the database.

Parameters:

  • table_name – The table to be dropped

  • commit – Boolean indicating if this dropping transaction should be committed

Returns:

None

static dt_now() → now[source]

Shorthand for returning a datetime.datetime.now() object :return: datetime.datetime.now()

static dt_today() → today[source]

Shorthand for returning a datetime.date.today() object :return: datetime.date.today()

encrypt_value(value: Any, encoding: str = 'utf-8') → bytes[source]

Users can pass a Fernet Encryption Key to the constructor of this class, possibly in the case that fields within the database are encrypted. Using this key, this method will encrypt a value. If no key is passed in the constructor (or not set after the fact) this method will raise an Exception.

Parameters:

  • value – The value to be encrypted, note that this value will be turned into a string and then bytes before encryption

  • encoding – The encoding of the value to bytes

Returns:

bytes - The encrypted value

static encrypt_value_static(fernet_encryption_key: bytes, value: Any, encoding: str = 'utf-8') → bytes[source]

This static method can be used directly if the user has an encryption key but would not like to start a database session

Parameters:

  • fernet_encryption_key – The encryption key

  • value – The value to be encrypted, note that this value will be turned into a string and then bytes before encryption

  • encoding – The encoding of the value to bytes

Returns:

bytes - The encrypted value

export_query_to_access(out_filepath: str, out_table_name: str, in_sql: str, in_parameters: dict | list | None = None, in_show_head: bool = False) → DataFrame | None[source]

Will export query results to an MS Access database, only if the query returns results.

Parameters:

  • out_filepath – The filepath of the new Access database to be created

  • out_table_name – The name of the table that will be filled with the resulting data

  • in_sql – The sql statement to be executed

  • in_parameters – The parameters associated with a parameterized query

  • in_show_head – Boolean indicating if the head of the DataFrame should print to the console

Returns:

DataFrame of the result or None

export_query_to_excel(out_filepath: str, out_table_name: str, in_sql: str, in_parameters: dict | list | None = None, in_show_head: bool = False) → DataFrame | None[source]

Will export query results to an MS Excel sheet, only if the query returns results.

Parameters:

  • out_filepath – The filepath of the new Excel sheet to be created

  • out_table_name – The name of the table that will be filled with the resulting data

  • in_sql – The sql statement to be executed

  • in_parameters – The parameters associated with a parameterized query

  • in_show_head – Boolean indicating if the head of the DataFrame should print to the console

Returns:

DataFrame of the result or None

export_query_to_sqlite(out_filepath: str, out_table_name: str, in_sql: str, in_parameters: dict | list | None = None, in_show_head: bool = False) → DataFrame | None[source]

Will export query results to an SQLite database, only if the query returns results.

Parameters:

  • out_filepath – The filepath of the new SQLite database to be created

  • out_table_name – The name of the table that will be filled with the resulting data

  • in_sql – The sql statement to be executed

  • in_parameters – The parameters associated with a parameterized query

  • in_show_head – Boolean indicating if the head of the DataFrame should print to the console

Returns:

DataFrame of the result or None

static flatten_sql(sql: str) → str[source]
classmethod generate_fernet_encryption_key() → bytes[source]

Will generate a Fernet Encryption Key. Note that if values are encrypted and inserted into the database using this key, then this key should be stored somewhere secure as this will be the only key which will correctly decrypt the values.

Returns:

bytes - The Fernet Encryption Key

init_table_schemas_from_files(create_table_sql_files_list: list, commit: bool = True) → None[source]

Will create tables within a BLANK database from files read in the create_table_sql_files_list. Note that if a database already contains tables, the method will not proceed and a warning will be printed to the console. For creating new tables within an existing database, use the query() method directly.

Parameters:

  • create_table_sql_files_list – A list of filepaths that contain the CREATE TABLE sql statements

  • commit – Should this transaction be committed

Returns:

None

init_table_schemas_from_sql(create_table_sql_statements_list: list, commit: bool = True) → None[source]

Will create tables within a BLANK database CREATE TABLE sql statements in the create_table_sql_statements_list. Note that if a database already contains tables, the method will not proceed and a warning will be printed to the console. For creating new tables within an existing database, use the query() method directly.

Parameters:

  • create_table_sql_statements_list – A list of the CREATE TABLE sql statements

  • commit – Should this transaction be committed

Returns:

None

insert_values(table_name: str, **column_value_pairs) → None[source]

This method can be used to insert values into a given table. The column_value_pairs keyword arguments can be used to set the column name with the respective value.

Parameters:

  • table_name – The name of the table to insert values

  • column_value_pairs – Keyword arguments that represent the column names and their respective values

Returns:

None

static isna(value: Any) → bool[source]

Checks if a value “isna” according to pandas or if the value is None

Parameters:

value – The value to be checked

Returns:

bool

query(sql: str, parameters: dict | list | None = None, show_head: bool = True, index: str | list | None = None, warn_is_none: bool = True) → DataFrame | None[source]

Executes any user defined sql statement and if this sql statement returns data such as from a SELECT statement, the resulting data will be returned as a pandas DataFrame. Parameterized sql statements are accepted as well.

Note that if a query does not return data (as in the case of a modification statement) OR the query returns zero results the return value will be None.

Parameters:

  • sql – The sql statement to be executed

  • parameters – The parameters associated with a parameterized query

  • show_head – Boolean indicating if the head of the resulting DataFrame should be printed to the console

  • index – Can be used to set the index of the resulting DataFrame

  • warn_is_none – Boolean indicating if a warning should be printed to the console when the query returns zero results

Returns:

DataFrame or None

rollback() → None[source]

Rolls back the database to its most recent state.

Returns:

None

static sha_hash(value: Any, sha_type: Literal['sha1', 'sha224', 'sha256', 'sha384', 'sha512'] = 'sha256', encoding: str = 'utf-8') → str[source]

Will hash a value given the “sha_type” hashing algorithm. Note that the default hashing algorithm is sha256 even if the “sha_type” argument isn’t valid. This could be useful for inserting passwords into a database.

Parameters:

  • value – The value to hash

  • sha_type – The hashing algorithm to use

  • encoding – The encoding of the value

Returns:

str - The “hexdigest” string of the hash

table_names(show_names: bool = False) → list[source]

Returns a list of the table names within the database.

Parameters:

show_names – Boolean indicating if the table names should be printed to the console

Returns:

list - The list of the table names

static uuid4()[source]

Returns a universal unique identifier (UUID) version 4. Can be useful for primary key generation, if that is the desired primary key strategy of the database.

Returns:

str - The uuid v4

dbpd.mysql module

class dbpd.mysql.MySQL(username: str, password: str, host: str, database_name: str, port: int | None = None, fernet_encryption_key: bytes | None = None, description: str = 'MySQL database connection with credentials', show_description: bool = True, **sqlalchemy_create_engine_kwargs)[source]

Bases: BaseDBPD

Using a SQLAlchemy session, it will connect to an MySQL database, methods can then be used from BaseDBPD.

The SQLAlchemy connection string is automatically generated from the init arguments.

Parameters:

  • username – The username of the database

  • password – The password of the database

  • host – The host location of the database

  • database_name – The name of the database

  • port – The port at which the database can be located

  • description – A description of the database (for reference utility only)

  • show_description – Whether the description should be printed to the console upon init

  • sqlalchemy_create_engine_kwargs – Any keyword arguments that should be passed to the sqlalchemy.create_engine() function

dbpd.oracle module

class dbpd.oracle.Oracle(username: str, password: str, host: str, sid: str = 'prod', port: int | None = None, threaded: bool = True, fernet_encryption_key: bytes | None = None, description: str = 'Oracle database connection with credentials', show_description: bool = True, **sqlalchemy_create_engine_kwargs)[source]

Bases: BaseDBPD

Using a SQLAlchemy session, it will connect to an Oracle database, methods can then be used from BaseDBPD.

The SQLAlchemy connection string is automatically generated from the init arguments.

Parameters:

  • username – The username of the database

  • password – The password of the database

  • host – The host location of the database

  • sid – The site identifier of the database

  • port – The port at which the database can be located

  • threaded – Should the connection session be threaded

  • description – A description of the database (for reference utility only)

  • show_description – Whether the description should be printed to the console upon init

  • sqlalchemy_create_engine_kwargs – Any keyword arguments that should be passed to the sqlalchemy.create_engine() function

dbpd.postgres module

class dbpd.postgres.Postgres(username: str, password: str, host: str, database_name: str, postgres_schema: str = 'public', port: int | None = None, fernet_encryption_key: bytes | None = None, description: str = 'Postgres database connection with credentials', show_description: bool = True, **sqlalchemy_create_engine_kwargs)[source]

Bases: BaseDBPD

Using a SQLAlchemy session, it will connect to an Postgres database, methods can then be used from BaseDBPD.

The SQLAlchemy connection string is automatically generated from the init arguments.

Parameters:

  • username – The username of the database

  • password – The password of the database

  • host – The host location of the database

  • database_name – The name of the database

  • postgres_schema – The schema of the database tables

  • port – The port at which the database can be located

  • description – A description of the database (for reference utility only)

  • show_description – Whether the description should be printed to the console upon init

  • sqlalchemy_create_engine_kwargs – Any keyword arguments that should be passed to the sqlalchemy.create_engine() function

dbpd.sqlite module

class dbpd.sqlite.SQLite(filepath: str, fernet_encryption_key: bytes | None = None, description: str = 'SQLite database connection', show_description: bool = True, **sqlite_connection_kwargs)[source]

Bases: BaseDBPD

Will connect to (or create) a SQLite database, methods can then be used from BaseDBPD

Parameters:

  • filepath – The filepath of the SQLite database (or where the new one should be created)

  • description – A description of the database (for reference utility only)

  • show_description – Whether the description should be printed to the console upon init

  • sqlite_connection_kwargs – Any keyword arguments that should be passed to the sqlite3.connect() function

class dbpd.sqlite.SQLiteInMemory(fernet_encryption_key: bytes | None = None, description: str = 'In-Memory SQLite database connection', show_description: bool = True, **sqlite_connection_kwargs)[source]

Bases: BaseDBPD

Will create an in-memory SQLite database, methods can then be used from BaseDBPD The first argument for sqlite3.connect() will be ‘:memory:’

Use the save_as() method to save the in-memory database to disk

Parameters:

  • description – A description of the database (for reference utility only)

  • show_description – Whether the description should be printed to the console upon init

  • sqlite_connection_kwargs – Any keyword arguments that should be passed to the sqlite3.connect() function

save_as(filepath: str, return_new_database_manager: bool = False) → SQLite | None[source]

Saves an in-memory SQLite database to a SQLite database file

Parameters:

  • filepath – The filepath of the new SQLite database

  • return_new_database_manager – Boolean indicating if the method should return a dbpd.SQLite object of the new database

Returns:

Optional[dbpd.SQLite]

Module contents