cardinal_pythonlib.rnc_db

---

Original code copyright (C) 2009-2022 Rudolf Cardinal (rudolf@pobox.com).

This file is part of cardinal_pythonlib.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

https://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

DEPRECATED support functions to interface Python to SQL-based databases conveniently.

DEFUNCT – use SQLAlchemy instead; it’s much better.

Not documented properly as it’s deprecated.

Regarding fieldspecs and fieldspec lists:

An individual fieldspec is a dictionary, e.g.

{ "name": "q1", "sqltype": "INTEGER" }

or

dict(name="q1", sqltype="INTEGER")

Possible keys are:

name: field name
sqltype: SQL type
notnull (optional): True for NOT NULL
autoincrement (optional): true for AUTO_INCREMENT
pk (optional): True for a PK
unique (optional): True for a UNIQUE (but not PK) field
comment (optional): string
value (optional): value for an individual record (not currently used)
indexed (optional): should the field be specifically indexed?
index_nchar (optional): specify if the field needs an index key length

General note about Python None/NULL handling:

# Fine:
cursor.execute("SELECT * FROM mytable WHERE myfield=?", 1)
# NOT fine; will return no rows:
cursor.execute("SELECT * FROM mytable WHERE myfield=?", None)
# Fine
cursor.execute("SELECT * FROM mytable WHERE myfield IS NULL")

JDBC types:

# https://webcache.googleusercontent.com/search?q=cache:WoMF0RGkqwgJ:www.tagwith.com/question_439319_jpype-and-jaydebeapi-returns-jpype-jclass-java-lang-long  # noqa

The object returned by JPype is a Python version of Java's java.lang.Long
class. To get the value out of it, use the value attribute:

n = java.lang.Long(44)
n
    <jpype._jclass.java.lang.Long object at 0x2377390>
n.value
    44L

JayDeBeApi contains a dict (_DEFAULT_CONVERTERS) that maps types it
recognises to functions that convert the Java values to Python values. This
dict can be found at the bottom of the dbapi2.pysource code. BIGINT is not
included in this dict, so objects of that database type don't get mapped
out of Java objects into Python values.

It's fairly easy to modify JayDeBeApi to add support for BIGINTs. Edit the
dbapi2.py file that contains most of the JayDeBeApi code and add the line

    'BIGINT': _java_to_py('longValue'),

to the _DEFAULT_CONVERTERS dict.

class cardinal_pythonlib.rnc_db.Access

classmethod describe_table(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str) → List[List[Any]]

Returns details on a specific table.

classmethod fetch_column_names(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str) → List[str]

Returns all column names for a table.

classmethod get_all_table_details(db: cardinal_pythonlib.rnc_db.DatabaseSupporter) → List[List[Any]]

Returns all information the database has on a table.

classmethod get_all_table_names(db: cardinal_pythonlib.rnc_db.DatabaseSupporter) → List[str]

Returns all table names in the database.

classmethod get_column_type(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str, column: str) → str

Returns database SQL datatype for a column, e.g. VARCHAR(50).

classmethod get_datatype(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str, column: str) → str

Returns database SQL datatype for a column: e.g. VARCHAR.

class cardinal_pythonlib.rnc_db.DatabaseSupporter

Support class for databases using pyodbc or MySQLdb.

add_column(tablename: str, fieldspec: Dict[str, str]) → int

Adds a column to an existing table.

change_column_if_table_exists(tablename: str, oldfieldname: str, newfieldname: str, newdef: str) → Optional[int]

Renames a column and alters its definition.

column_exists(tablename: str, column: str) → bool

Does the column exist?

commit() → None

Commits the transaction.

connect(engine: str = None, interface: str = None, host: str = None, port: int = None, database: str = None, driver: str = None, dsn: str = None, odbc_connection_string: str = None, user: str = None, password: str = None, autocommit: bool = True, charset: str = 'utf8', use_unicode: bool = True) → bool

engine: access, mysql, sqlserver interface: mysql, odbc, jdbc

connect_to_database_odbc_access(dsn: str, autocommit: bool = True) → None

Connects to an Access database via ODBC, with the DSN prespecified.

connect_to_database_odbc_mysql(database: str, user: str, password: str, server: str = 'localhost', port: int = 3306, driver: str = '{MySQL ODBC 5.1 Driver}', autocommit: bool = True) → None

Connects to a MySQL database via ODBC.

connect_to_database_odbc_sqlserver(odbc_connection_string: str = None, dsn: str = None, database: str = None, user: str = None, password: str = None, server: str = 'localhost', driver: str = '{SQL Server}', autocommit: bool = True) → None

Connects to an SQL Server database via ODBC.

count_where(table: str, wheredict: Dict[str, Any] = None) → int

Counts rows in a table, given a set of WHERE criteria (ANDed), returning a count.

create_fulltext_index(table: str, field: str, indexname: str = None) → Optional[int]

Creates a FULLTEXT index (default name _idxft_FIELDNAME), unless it exists already. See:

https://dev.mysql.com/doc/refman/5.6/en/innodb-fulltext-index.html https://dev.mysql.com/doc/refman/5.0/en/fulltext-search.html

create_index(table: str, field: str, nchars: int = None, indexname: str = None, unique: bool = False) → Optional[int]

Creates an index (default name _idx_FIELDNAME), unless it exists already.

create_index_from_fieldspec(table: str, fieldspec: Dict[str, str], indexname: str = None) → None

Calls create_index based on a fieldspec, if the fieldspec has indexed = True.

create_or_replace_primary_key(table: str, fieldnames: Sequence[str]) → int

Make a primary key, or replace it if it exists.

create_or_update_table(tablename: str, fieldspeclist: List[Dict[str, str]], drop_superfluous_columns: bool = False, dynamic: bool = False, compressed: bool = False) → None

• Make table, if it doesn’t exist.
• Add fields that aren’t there.
• Warn about superfluous fields, but don’t delete them, unless drop_superfluous_columns == True.
• Make indexes, if requested.

cursor()

Returns database cursor, or raises NoDatabaseError.

db_exec(sql: str, *args) → int

Executes SQL (with “?” placeholders for arguments).

db_exec_and_commit(sql: str, *args) → int

Execute SQL and commit.

db_exec_literal(sql: str) → int

Executes SQL without modification. Returns rowcount.

db_exec_with_cursor(cursor, sql: str, *args) → int

Executes SQL on a supplied cursor, with “?” placeholders, substituting in the arguments. Returns number of rows affected.

debug_query(sql: str, *args) → None

Executes SQL and writes the result to the log.

delete_by_field(table: str, field: str, value: Any) → int

Deletes all records where “field” is “value”.

delimit(x: str) → str

Delimits e.g. a fieldname.

describe_table(table: str) → List[List[Any]]

Returns details on a specific table.

does_row_exist(table: str, field: str, value: Any) → bool

Checks for the existence of a record by a single field (typically a PK).

drop_column(tablename: str, fieldname: str) → int

Drops (deletes) a column from an existing table.

drop_table(tablename: str) → int

Drops a table. Use caution!

drop_view(viewname: str) → int

Drops a view.

ensure_db_open() → None

Raise NoDatabaseError if the database isn’t open.

fetch_all_objects_from_db(cls: Type[T], table: str, fieldlist: Sequence[str], construct_with_pk: bool, *args) → List[T]

Fetches all objects from a table, returning an array of objects of class cls.

fetch_all_objects_from_db_by_pklist(cls: Type[CT_co], table: str, fieldlist: Sequence[str], pklist: Sequence[Any], construct_with_pk: bool, *args) → List[T]

Fetches all objects from a table, given a list of PKs.

fetch_all_objects_from_db_where(cls: Type[T], table: str, fieldlist: Sequence[str], construct_with_pk: bool, wheredict: Optional[Dict[str, Any]], *args) → List[T]

Fetches all objects from a table, given a set of WHERE criteria (ANDed), returning an array of objects of class cls. As usual here, the first field in the fieldlist must be the PK.

fetch_column_names(table: str) → List[str]

Returns all column names for a table.

fetch_fieldnames(sql: str, *args) → List[str]

Executes SQL; returns just the output fieldnames.

fetch_object_from_db_by_other_field(obj: Any, table: str, fieldlist: Sequence[str], keyname: str, keyvalue: Any) → bool

Fetches object from database table by a field specified by keyname/keyvalue. Writes back to object. Returns True/False for success/failure.

fetch_object_from_db_by_pk(obj: Any, table: str, fieldlist: Sequence[str], pkvalue: Any) → bool

Fetches object from database table by PK value. Writes back to object. Returns True/False for success/failure.

fetchall(sql: str, *args) → Sequence[Sequence[Any]]

Executes SQL; returns all rows, or [].

fetchall_as_dictlist(sql: str, *args) → List[Dict[str, Any]]

Executes SQL; returns list of dictionaries, where each dict contains fieldname/value pairs.

fetchall_with_fieldnames(sql: str, *args) → Tuple[Sequence[Sequence[Any]], Sequence[str]]

Executes SQL; returns (rows, fieldnames).

fetchallfirstvalues(sql: str, *args) → List[Any]

Executes SQL; returns list of first values of each row.

fetchone(sql: str, *args) → Optional[Sequence[Any]]

Executes SQL; returns the first row, or None.

fetchvalue(sql: str, *args) → Optional[Any]

Executes SQL; returns the first value of the first row, or None.

static fielddefsql_from_fieldspec(fieldspec: Dict[str, str]) → str

Returns SQL fragment to define a field.

fielddefsql_from_fieldspeclist(fieldspeclist: List[Dict[str, str]]) → str

Returns list of field-defining SQL fragments.

static fieldname_from_fieldspec(fieldspec: Dict[str, str]) → str

Returns a fieldname from a field specification.

static fieldnames_from_fieldspeclist(fieldspeclist: List[Dict[str, str]]) → List[str]

Returns fieldnames from a field specification list.

static fieldspec_subset_by_name(fieldspeclist: List[Dict[str, str]], fieldnames: Container[str]) → List[Dict[str, str]]

Returns a subset of the fieldspecs matching the fieldnames list.

gen_fetchall(sql: str, *args) → Iterator[Sequence[Any]]

fetchall() as a generator.

gen_fetchfirst(sql: str, *args) → Iterator[Any]

fetch first values, as a generator.

get_all_table_details() → List[List[Any]]

Returns all information the database has on a table.

get_all_table_names() → List[str]

Returns all table names in the database.

get_column_type(table: str, column: str) → str

Returns database SQL datatype for a column, e.g. VARCHAR(50).

get_comment(table: str, column: str) → str

Returns database SQL comment for a column.

get_datatype(table: str, column: str) → str

Returns database SQL datatype for a column: e.g. VARCHAR.

index_exists(table: str, indexname: str) → bool

Does an index exist? (Specific to MySQL.)

insert_multiple_records(table: str, fields: Sequence[str], records: Sequence[Sequence[Any]]) → int

Inserts a record into database, table “table”, using the list of fieldnames and the list of records (each a list of values). Returns number of rows affected.

insert_object_into_db_pk_known(obj: Any, table: str, fieldlist: Sequence[str]) → None

Inserts object into database table, with PK (first field) already known.

insert_object_into_db_pk_unknown(obj: Any, table: str, fieldlist: Sequence[str]) → None

Inserts object into database table, with PK (first field) initially unknown (and subsequently set in the object from the database).

insert_record(table: str, fields: Sequence[str], values: Sequence[Any], update_on_duplicate_key: bool = False) → int

Inserts a record into database, table “table”, using the list of fieldnames and the list of values. Returns the new PK (or None).

insert_record_by_dict(table: str, valuedict: Dict[str, Any]) → Optional[int]

Inserts a record into database, table “table”, using a dictionary containing field/value mappings. Returns the new PK (or None).

insert_record_by_fieldspecs_with_values(table: str, fieldspeclist: List[Dict[str, str]]) → int

Inserts a record into the database using a list of fieldspecs having their value stored under the ‘value’ key.

is_open() → bool

Is the database open?

is_read_only() → bool

Does the user have read-only access to the database? This is a safety check, but should NOT be the only safety check!

localize_sql(sql: str) → str

Translates ?-placeholder SQL to appropriate dialect.

For example, MySQLdb uses %s rather than ?.

make_table(tablename: str, fieldspeclist: List[Dict[str, str]], dynamic: bool = False, compressed: bool = False) → Optional[int]

Makes a table, if it doesn’t already exist.

modify_column_if_table_exists(tablename: str, fieldname: str, newdef: str) → Optional[int]

Alters a column’s definition without renaming it.

ping() → None

Pings a database connection, reconnecting if necessary.

rename_table(from_table: str, to_table: str) → Optional[int]

Renames a table. MySQL-specific.

rollback() → None

Rolls back the transaction.

save_object_to_db(obj: Any, table: str, fieldlist: Sequence[str], is_new_record: bool) → None

Saves a object to the database, inserting or updating as necessary.

table_exists(tablename: str) → bool

Does the table exist?

update_object_in_db(obj: Any, table: str, fieldlist: Sequence[str]) → None

Updates an object in the database (saves it to the database, where it exists there already).

wipe_table(table: str) → int

Delete all records from a table. Use caution!

class cardinal_pythonlib.rnc_db.Flavour

Describes a database “flavour” (dialect).

classmethod describe_table(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str) → List[List[Any]]

Returns details on a specific table.

classmethod fetch_column_names(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str) → List[str]

Returns all column names for a table.

classmethod get_all_table_details(db: cardinal_pythonlib.rnc_db.DatabaseSupporter) → List[List[Any]]

Returns all information the database has on a table.

classmethod get_all_table_names(db: cardinal_pythonlib.rnc_db.DatabaseSupporter) → List[str]

Returns all table names in the database.

classmethod get_column_type(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str, column: str) → str

Returns database SQL datatype for a column, e.g. VARCHAR(50).

classmethod get_comment(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str, column: str) → str

Returns database SQL comment for a column.

classmethod get_datatype(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str, column: str) → str

Returns database SQL datatype for a column: e.g. VARCHAR.

classmethod get_system_variable(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, varname: str) → Any

Returns a database system variable.

class cardinal_pythonlib.rnc_db.MySQL

classmethod describe_table(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str) → List[List[Any]]

Returns details on a specific table.

classmethod fetch_column_names(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str) → List[str]

Returns all column names for a table.

classmethod get_all_table_details(db: cardinal_pythonlib.rnc_db.DatabaseSupporter) → List[List[Any]]

Returns all information the database has on a table.

classmethod get_all_table_names(db: cardinal_pythonlib.rnc_db.DatabaseSupporter) → List[str]

Returns all table names in the database.

classmethod get_column_type(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str, column: str) → str

Returns database SQL datatype for a column, e.g. VARCHAR(50).

classmethod get_comment(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str, column: str) → str

Returns database SQL comment for a column.

classmethod get_datatype(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str, column: str) → str

Returns database SQL datatype for a column: e.g. VARCHAR.

classmethod get_system_variable(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, varname: str) → Any

Returns a database system variable.

classmethod is_read_only(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, logger: logging.Logger = None) → bool

Do we have read-only access?

exception cardinal_pythonlib.rnc_db.NoDatabaseError(value: str = '')

Exception class for when a database is unavailable.

class cardinal_pythonlib.rnc_db.SQLServer

classmethod describe_table(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str) → List[List[Any]]

Returns details on a specific table.

classmethod fetch_column_names(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str) → List[str]

Returns all column names for a table.

classmethod get_all_table_details(db: cardinal_pythonlib.rnc_db.DatabaseSupporter) → List[List[Any]]

Returns all information the database has on a table.

classmethod get_all_table_names(db: cardinal_pythonlib.rnc_db.DatabaseSupporter) → List[str]

Returns all table names in the database.

classmethod get_column_type(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str, column: str) → str

Returns database SQL datatype for a column, e.g. VARCHAR(50).

classmethod get_datatype(db: cardinal_pythonlib.rnc_db.DatabaseSupporter, table: str, column: str) → str

Returns database SQL datatype for a column: e.g. VARCHAR.

cardinal_pythonlib.rnc_db.add_master_user_mysql(database: str, root_user: str, root_password: str, new_user: str, new_password: str, server: str = 'localhost', port: int = 3306, charset: str = 'utf8', use_unicode: bool = True, localhost_only: bool = True) → None

Connects via PyMySQL/MySQLdb and creates a database superuser.

cardinal_pythonlib.rnc_db.assign_from_list(obj: T, fieldlist: Sequence[str], valuelist: Sequence[any]) → None

Within “obj”, assigns the values from the value list to the fields in the fieldlist.

cardinal_pythonlib.rnc_db.blank_object(obj: T, fieldlist: Sequence[str]) → None

Within “obj”, sets all fields in the fieldlist to None.

cardinal_pythonlib.rnc_db.create_database_mysql(database: str, user: str, password: str, server: str = 'localhost', port: int = 3306, charset: str = 'utf8', collate: str = 'utf8_general_ci', use_unicode: bool = True) → bool

Connects via PyMySQL/MySQLdb and creates a database.

cardinal_pythonlib.rnc_db.create_object_from_list(cls: Type[CT_co], fieldlist: Sequence[str], valuelist: Sequence[Any], *args, **kwargs) → T

Create an object by instantiating cls(*args, **kwargs) and assigning the values in valuelist to the fields in fieldlist.

If construct_with_pk is True, initialize with cls(valuelist[0], *args, **kwargs) and assign the values in valuelist[1:] to fieldlist[1:].

Note: in Python 3, we could define as

...(... valuelist, *args, construct_with_pk=False, **kwargs):

but not in Python 2, and this is meant to be back-compatible.

cardinal_pythonlib.rnc_db.datetime2literal_rnc(d: datetime.datetime, c: Optional[Dict[KT, VT]]) → str

Format a DateTime object as something MySQL will actually accept.

cardinal_pythonlib.rnc_db.debug_object(obj: T) → str

Prints key/value pairs for an object’s dictionary.

cardinal_pythonlib.rnc_db.debug_query_result(rows: Sequence[Any]) → None

Writes a query result to the log.

cardinal_pythonlib.rnc_db.debug_sql(sql: str, *args) → None

Writes SQL and arguments to the log.

cardinal_pythonlib.rnc_db.delimit(x: str, delims: Tuple[str, str]) → str

Delimits x, using delims[0] (left) and delims[1] (right).

cardinal_pythonlib.rnc_db.dump_database_object(obj: T, fieldlist: Iterable[str]) → None

Prints key/value pairs for an object’s dictionary.

cardinal_pythonlib.rnc_db.full_datatype_to_mysql(d: str) → str

Converts a full datatype, e.g. INT, VARCHAR(10), VARCHAR(MAX), to a MySQL equivalent.

cardinal_pythonlib.rnc_db.get_pk_of_last_insert(cursor) → int

Returns the primary key of the last insert performed with the cursor.

cardinal_pythonlib.rnc_db.get_sql_insert(table: str, fieldlist: Sequence[str], delims: Tuple[str, str] = ('', '')) → str

Returns ?-marked SQL for an INSERT statement.

cardinal_pythonlib.rnc_db.get_sql_insert_or_update(table: str, fieldlist: Sequence[str], delims: Tuple[str, str] = ('', '')) → str

Returns ?-marked SQL for an INSERT-or-if-duplicate-key-UPDATE statement.

cardinal_pythonlib.rnc_db.get_sql_insert_without_first_field(table: str, fieldlist: Sequence[str], delims: Tuple[str, str] = ('', '')) → str

Returns ?-marked SQL for an INSERT statement, ignoring the first field (typically, the PK).

cardinal_pythonlib.rnc_db.get_sql_select_all_fields_by_key(table: str, fieldlist: Sequence[str], keyname: str, delims: Tuple[str, str] = ('', '')) → str

Returns SQL: SELECT [all fields in the fieldlist] WHERE [keyname] = ?

cardinal_pythonlib.rnc_db.get_sql_select_all_non_pk_fields_by_pk(table: str, fieldlist: Sequence[str], delims: Tuple[str, str] = ('', '')) → str

Returns SQL: SELECT [all but the first field] WHERE [the first field] = ?

cardinal_pythonlib.rnc_db.get_sql_update_by_first_field(table: str, fieldlist: Sequence[str], delims: Tuple[str, str] = ('', '')) → str

Returns SQL for an UPDATE statement, to update all fields except the first field (PK) using the PK as the key.

cardinal_pythonlib.rnc_db.set_verbose_logging(verbose: bool) → None

Chooses basic or verbose logging.

cardinal_pythonlib.rnc_db.sql_dequote_string(s: str) → str

Reverses sql_quote_string.

cardinal_pythonlib.rnc_db.sql_quote_string(s: str) → str

Quotes string, escaping apostrophes by doubling them.