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
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.
- 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:
Restrictions:
Specify either
sqla_column
ormultiple_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 listb
, as percolumns_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
(seecolumn_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 toTrue
(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 ofJones
, whereJones
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 leastmin_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
andTEXT
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 (orNone
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
, seecardinal_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’sAUTOINCREMENT
.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;
Also:
sp_columns 'tablename';
… which is what SQLAlchemy does (
dialects/mssql/base.py
, inget_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 SQLAlchemyInteger(length=11)
.NOTE that the reverse operation is performed by
str(coltype)
orcoltype.compile()
orcoltype.compile(dialect)
; seeTypeEngine
.DIALECT-AWARE.
- Parameters:
- 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 classa dialect contains these things of interest:
ischema_names
: string-to-class dictionarytype_compiler
: instance of e.g.sqlalchemy.sql.compiler.GenericTypeCompiler
. This has aprocess()
method, but that operates onTypeEngine
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.htmlAn 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 itssqlalchemy.sql.sqltypes.INTEGER
class takes no parameters, so you get the errorTypeError: object() takes no parameters
. Similarly, MySQL’sDATETIME(6)
uses the 6 to refer to precision, but theDATETIME
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 listb
, as perindexes_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 asFloat
,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 aBigInteger
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). ReturnsNone
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?