cardinal_pythonlib.sqlalchemy.schema

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.

Functions to work with SQLAlchemy schemas (schemata) directly, via SQLAlchemy Core.

Functions that have to work with specific dialect information are marked DIALECT-AWARE.

class cardinal_pythonlib.sqlalchemy.schema.SqlaColumnInspectionInfo(sqla_info_dict: Dict[str, Any])[source]

Class to represent information from inspecting a database column.

A clearer way of getting information than the plain dict that SQLAlchemy uses.

Parameters:

sqla_info_dict¶ –

see

cardinal_pythonlib.sqlalchemy.schema.add_index(engine: Engine, sqla_column: Column | None = None, multiple_sqla_columns: List[Column] | None = None, unique: bool = False, fulltext: bool = False, length: int | None = None) → None[source]

Adds an index to a database column (or, in restricted circumstances, several columns).

The table name is worked out from the Column object.

DIALECT-AWARE.

Parameters:

engine¶ – SQLAlchemy Engine object
sqla_column¶ – single column to index
multiple_sqla_columns¶ – multiple columns to index (see below)
unique¶ – make a UNIQUE index?
fulltext¶ – make a FULLTEXT index?
length¶ – index length to use (default None)

Restrictions:

Specify either sqla_column or multiple_sqla_columns, not both.
The normal method is sqla_column.
multiple_sqla_columns is only used for Microsoft SQL Server full-text indexing (as this database permits only one full-text index per table, though that index can be on multiple columns).

cardinal_pythonlib.sqlalchemy.schema.column_creation_ddl(sqla_column: Column, dialect: Dialect) → str[source]

Returns DDL to create a column, using the specified dialect.

The column should already be bound to a table (because e.g. the SQL Server dialect requires this for DDL generation). If you don’t append the column to a Table object, the DDL generation step gives “sqlalchemy.exc.CompileError: mssql requires Table-bound columns in order to generate DDL”.

Testing: see schema_tests.py

cardinal_pythonlib.sqlalchemy.schema.column_lists_equal(a: List[Column], b: List[Column]) → bool[source]: Are all columns in list a equal to their counterparts in list b, as per columns_equal()?

cardinal_pythonlib.sqlalchemy.schema.column_types_equal(a_coltype: TypeEngine, b_coltype: TypeEngine) → bool[source]

Checks that two SQLAlchemy column types are equal (by comparing str() versions of them).

See https://stackoverflow.com/questions/34787794/sqlalchemy-column-type-comparison.

IMPERFECT.

cardinal_pythonlib.sqlalchemy.schema.columns_equal(a: Column, b: Column) → bool[source]

Are two SQLAlchemy columns are equal? Checks based on:

column name
column type (see column_types_equal())
nullable

cardinal_pythonlib.sqlalchemy.schema.convert_sqla_type_for_dialect(coltype: TypeEngine, dialect: Dialect, strip_collation: bool = True, convert_mssql_timestamp: bool = True, expand_for_scrubbing: bool = False) → TypeEngine[source]

Converts an SQLAlchemy column type from one SQL dialect to another.

DIALECT-AWARE.

Parameters:

coltype¶ – SQLAlchemy column type in the source dialect
dialect¶ – destination Dialect
strip_collation¶ – remove any COLLATION information?
convert_mssql_timestamp¶ – since you cannot write to a SQL Server TIMESTAMP field, setting this option to True (the default) converts such types to something equivalent but writable.
expand_for_scrubbing¶ – The purpose of expand_for_scrubbing is that, for example, a VARCHAR(200) field containing one or more instances of Jones, where Jones is to be replaced with [XXXXXX], will get longer (by an unpredictable amount). So, better to expand to unlimited length.

Returns:

an SQLAlchemy column type instance, in the destination dialect

cardinal_pythonlib.sqlalchemy.schema.does_sqlatype_merit_fulltext_index(coltype: TypeEngine | Type[Visitable], min_length: int = 1000) → bool[source]: Is the SQLAlchemy column type a type that might merit a FULLTEXT index (meaning a string type of at least min_length)?

cardinal_pythonlib.sqlalchemy.schema.does_sqlatype_require_index_len(coltype: TypeEngine | Type[Visitable]) → bool[source]

Is the SQLAlchemy column type one that requires its indexes to have a length specified?

(MySQL, at least, requires index length to be specified for BLOB and TEXT columns: https://dev.mysql.com/doc/refman/5.7/en/create-index.html.)

cardinal_pythonlib.sqlalchemy.schema.execute_ddl(engine: Engine, sql: str | None = None, ddl: ExecutableDDLElement | None = None) → None[source]

Execute DDL, either from a plain SQL string, or from an SQLAlchemy DDL element.

Previously we would use DDL(sql, bind=engine).execute(), but this has gone in SQLAlchemy 2.0.

Note that creating the DDL object with e.g. ddl = DDL(sql).execute_if(dialect=SqlaDialectName.SQLSERVER), and passing that DDL object to this function, does NOT make execution condition; it executes regardless. The execute_if() construct is used for listeners; see https://docs.sqlalchemy.org/en/20/core/ddl.html#sqlalchemy.schema.ExecutableDDLElement.execute_if

cardinal_pythonlib.sqlalchemy.schema.gen_columns_info(engine: Engine, tablename: str) → Generator[SqlaColumnInspectionInfo, None, None][source]: For the specified table, generate column information as SqlaColumnInspectionInfo objects.

cardinal_pythonlib.sqlalchemy.schema.get_column_info(engine: Engine, tablename: str, columnname: str) → SqlaColumnInspectionInfo | None[source]: For the specified column in the specified table, get column information as a SqlaColumnInspectionInfo object (or None if such a column can’t be found).

cardinal_pythonlib.sqlalchemy.schema.get_column_names(engine: Engine, tablename: str) → List[str][source]: Get all the database column names for the specified table.

cardinal_pythonlib.sqlalchemy.schema.get_column_type(engine: Engine, tablename: str, columnname: str) → TypeEngine | None[source]

For the specified column in the specified table, get its type as an instance of an SQLAlchemy column type class (or None if such a column can’t be found).

For more on TypeEngine, see cardinal_pythonlib.orm_inspect.coltype_as_typeengine().

cardinal_pythonlib.sqlalchemy.schema.get_effective_int_pk_col(table_: Table) → str | None[source]: If a table has a single integer primary key, or a single integer AUTOINCREMENT column, return its column name; otherwise, None.

cardinal_pythonlib.sqlalchemy.schema.get_list_of_sql_string_literals_from_quoted_csv(x: str) → List[str][source]: Used to extract SQL column type parameters. For example, MySQL has column types that look like ENUM('a', 'b', 'c', 'd'). This function takes the "'a', 'b', 'c', 'd'" and converts it to ['a', 'b', 'c', 'd'].

cardinal_pythonlib.sqlalchemy.schema.get_pk_colnames(table_: Table) → List[str][source]: If a table has a PK, this will return its database column name(s); otherwise, None.

cardinal_pythonlib.sqlalchemy.schema.get_single_int_autoincrement_colname(table_: Table) → str | None[source]

If a table has a single integer AUTOINCREMENT column, this will return its name; otherwise, None.

It’s unlikely that a database has >1 AUTOINCREMENT field anyway, but we should check.
SQL Server’s IDENTITY keyword is equivalent to MySQL’s AUTOINCREMENT.

Verify against SQL Server:

SELECT table_name, column_name
FROM information_schema.columns
WHERE COLUMNPROPERTY(OBJECT_ID(table_schema + '.' + table_name),
                     column_name,
                     'IsIdentity') = 1
ORDER BY table_name;

… https://stackoverflow.com/questions/87747

Also:
```
sp_columns 'tablename';
```
… which is what SQLAlchemy does (dialects/mssql/base.py, in get_columns()).

cardinal_pythonlib.sqlalchemy.schema.get_single_int_pk_colname(table_: Table) → str | None[source]

If a table has a single-field (non-composite) integer PK, this will return its database column name; otherwise, None.

Note that it is legitimate for a database table to have both a composite primary key and a separate IDENTITY (AUTOINCREMENT) integer field. This function won’t find such columns.

cardinal_pythonlib.sqlalchemy.schema.get_sqla_coltype_from_dialect_str(coltype: str, dialect: Dialect) → TypeEngine[source]

Returns an SQLAlchemy column type, given a column type name (a string) and an SQLAlchemy dialect. For example, this might convert the string INTEGER(11) to an SQLAlchemy Integer(length=11).

NOTE that the reverse operation is performed by str(coltype) or coltype.compile() or coltype.compile(dialect); see TypeEngine.

DIALECT-AWARE.

Parameters:

dialect¶ – a SQLAlchemy Dialect class
coltype¶ – a str() representation, e.g. from str(c['type']) where c is an instance of sqlalchemy.sql.schema.Column.

Returns:

a Python object that is a subclass of sqlalchemy.types.TypeEngine

Example:

get_sqla_coltype_from_string('INTEGER(11)', engine.dialect)
# gives: Integer(length=11)

Notes:

sqlalchemy.engine.default.DefaultDialect is the dialect base class
a dialect contains these things of interest:
- ischema_names: string-to-class dictionary
- type_compiler: instance of e.g. sqlalchemy.sql.compiler.GenericTypeCompiler. This has a process() method, but that operates on TypeEngine objects.
- get_columns: takes a table name, inspects the database
example of the dangers of eval: https://nedbatchelder.com/blog/201206/eval_really_is_dangerous.html
An example of a function doing the reflection/inspection within SQLAlchemy is sqlalchemy.dialects.mssql.base.MSDialect.get_columns(), which has this lookup: coltype = self.ischema_names.get(type, None)

Caveats:

the parameters, e.g. DATETIME(6), do NOT necessarily either work at all or work correctly. For example, SQLAlchemy will happily spit out 'INTEGER(11)' but its sqlalchemy.sql.sqltypes.INTEGER class takes no parameters, so you get the error TypeError: object() takes no parameters. Similarly, MySQL’s DATETIME(6) uses the 6 to refer to precision, but the DATETIME class in SQLAlchemy takes only a boolean parameter (timezone).
However, sometimes we have to have parameters, e.g. VARCHAR length.
Thus, this is a bit useless.
Fixed, with a few special cases.

cardinal_pythonlib.sqlalchemy.schema.get_table_names(engine: Engine) → List[str][source]: Returns a list of database table names from the Engine.

cardinal_pythonlib.sqlalchemy.schema.get_view_names(engine: Engine) → List[str][source]: Returns a list of database view names from the Engine.

cardinal_pythonlib.sqlalchemy.schema.giant_text_sqltype(dialect: Dialect) → str[source]

Returns the SQL column type used to make very large text columns for a given dialect.

DIALECT-AWARE.

Parameters:: dialect¶ – a SQLAlchemy Dialect
Returns:: the SQL data type of “giant text”, typically ‘LONGTEXT’ for MySQL and ‘NVARCHAR(MAX)’ for SQL Server.

cardinal_pythonlib.sqlalchemy.schema.index_exists(engine: Engine, tablename: str, indexname: str | None = None, colnames: str | List[str] | None = None, raise_if_nonexistent_table: bool = True) → bool[source]

Does the specified index exist for the specified table?

You can specify either the name of the index, or the name(s) of columns. But not both.

If the table doesn’t exist, then if raise_if_nonexistent_table is True, raise sqlalchemy.exc.NoSuchTableError; otherwise, warn and return False.

cardinal_pythonlib.sqlalchemy.schema.index_lists_equal(a: List[Index], b: List[Index]) → bool[source]: Are all indexes in list a equal to their counterparts in list b, as per indexes_equal()?

cardinal_pythonlib.sqlalchemy.schema.indexes_equal(a: Index, b: Index) → bool[source]: Are two indexes equal? Checks by comparing str() versions of them. (AM UNSURE IF THIS IS ENOUGH.)

cardinal_pythonlib.sqlalchemy.schema.is_int_autoincrement_column(c: Column, t: Table) → bool[source]: Is this an integer AUTOINCREMENT column? Used by get_single_int_autoincrement_colname(); q.v.

cardinal_pythonlib.sqlalchemy.schema.is_sqlatype_binary(coltype: TypeEngine | Type[Visitable]) → bool[source]: Is the SQLAlchemy column type a binary type?

cardinal_pythonlib.sqlalchemy.schema.is_sqlatype_date(coltype: TypeEngine | Type[Visitable]) → bool[source]: Is the SQLAlchemy column type a date type?

cardinal_pythonlib.sqlalchemy.schema.is_sqlatype_integer(coltype: TypeEngine | Type[Visitable]) → bool[source]: Is the SQLAlchemy column type an integer type?

cardinal_pythonlib.sqlalchemy.schema.is_sqlatype_numeric(coltype: TypeEngine | Type[Visitable]) → bool[source]

Is the SQLAlchemy column type one that inherits from Numeric, such as Float, Decimal?

Note that integers don’t count as Numeric!

cardinal_pythonlib.sqlalchemy.schema.is_sqlatype_string(coltype: TypeEngine | Type[Visitable]) → bool[source]: Is the SQLAlchemy column type a string type?

cardinal_pythonlib.sqlalchemy.schema.is_sqlatype_text_of_length_at_least(coltype: TypeEngine | Type[Visitable], min_length: int = 1000) → bool[source]: Is the SQLAlchemy column type a string type that’s at least the specified length?

cardinal_pythonlib.sqlalchemy.schema.is_sqlatype_text_over_one_char(coltype: TypeEngine | Type[Visitable]) → bool[source]: Is the SQLAlchemy column type a string type that’s more than one character long?

cardinal_pythonlib.sqlalchemy.schema.make_bigint_autoincrement_column(column_name: str, nullable: bool = False, comment: str | None = None) → Column[source]: Returns an instance of Column representing a BigInteger AUTOINCREMENT column, or the closest that the database engine can manage.

cardinal_pythonlib.sqlalchemy.schema.mssql_get_pk_index_name(engine: Engine, tablename: str, schemaname: str = 'dbo') → str[source]: For Microsoft SQL Server specifically: fetch the name of the PK index for the specified table (in the specified schema), or '' if none is found.

cardinal_pythonlib.sqlalchemy.schema.mssql_table_has_ft_index(engine: Engine, tablename: str, schemaname: str = 'dbo') → bool[source]: For Microsoft SQL Server specifically: does the specified table (in the specified schema) have at least one full-text index?

cardinal_pythonlib.sqlalchemy.schema.mssql_transaction_count(engine_or_conn: Connection | Engine) → int[source]: For Microsoft SQL Server specifically: fetch the value of the TRANCOUNT variable (see e.g. https://docs.microsoft.com/en-us/sql/t-sql/functions/trancount-transact-sql?view=sql-server-2017). Returns None if it can’t be found (unlikely?).

cardinal_pythonlib.sqlalchemy.schema.remove_collation(coltype: TypeEngine) → TypeEngine[source]: Returns a copy of the specific column type with any COLLATION removed.

cardinal_pythonlib.sqlalchemy.schema.table_exists(engine: Engine, tablename: str) → bool[source]: Does the named table exist in the database?

cardinal_pythonlib.sqlalchemy.schema.table_or_view_exists(engine: Engine, table_or_view_name: str) → bool[source]: Does the named table/view exist (either as a table or as a view) in the database?

cardinal_pythonlib.sqlalchemy.schema.view_exists(engine: Engine, viewname: str) → bool[source]: Does the named view exist in the database?