使用 mssql-python 进行模式发现

mssql-python 光标类提供了九种元数据方法,这些方法映射到 ODBC 目录函数。 利用这些方法以程序方式发现表、列、存储过程、键和索引。 它们帮助你构建数据驱动的应用,能够在运行时适应数据库模式,比如迁移工具、代码生成器或管理仪表盘。

方法 ODBC 函数 Returns 何时使用
tables() SQLTables 表格和视图信息。 库存数据库。 在查询前验证表的存在性。
columns() SQLColumns 列详细信息。 生成DDL,构建动态查询,或将列映射到代码中。
procedures() SQLProcedures 存储过程信息。 发现可用的API。 生成过程调用封装器。
primaryKeys() SQLPrimaryKeys 主键列。 识别用于 UPDATE/DELETE 操作的唯一行标识符。
foreignKeys() SQLForeignKeys 外国密钥关系。 梳理表之间的关系,确定清理脚本的删除顺序。
statistics() SQLStatistics 索引和统计信息。 验证指数覆盖范围以进行性能调优。
rowIdColumns() SQLSpecialColumns (ROWID) 唯一行标识列。 查找最适合用于识别特定行的列。
rowVerColumns() SQLSpecialColumns (ROWVER) 行版本列。 实现乐观并发控制(检测并发修改)。
getTypeInfo() SQLGetTypeInfo 数据类型信息。 发现支持的跨平台兼容性类型。

每个方法都会返回一个光标,你可以遍历以访问结果。

Tables

数据库中的表和视图列表:

cursor = conn.cursor()

# List all tables
for row in cursor.tables():
    print(f"{row.table_schem}.{row.table_name} ({row.table_type})")

# Filter by name (supports wildcards % and _)
for row in cursor.tables(table="Product%"):
    print(row.table_name)

# Filter by schema
for row in cursor.tables(schema="Sales"):
    print(row.table_name)

# Filter by type
for row in cursor.tables(tableType="TABLE"):  # Excludes views
    print(row.table_name)

tables() 参数

以下参数用于控制表发现:

参数 描述
table 表名模式(支持 %_ 通配符)。
catalog 目录(数据库)名称。
schema 架构名称模式。
tableType 按类型筛选:TABLEVIEWSYSTEM TABLEGLOBAL TEMPORARYLOCAL TEMPORARYALIAS, , 。 SYNONYM

tables() 结果列

tables() 方法为每个表或视图返回以下列:

描述
table_cat 目录(数据库)名称。
table_schem 架构名称。
table_name 表或视图名称。
table_type TABLEVIEWSYSTEM TABLEGLOBAL TEMPORARYLOCAL TEMPORARYALIAS, , 。 SYNONYM
remarks 描述或评论。

检查表是否存在

在查询表之前,先验证表的存在:

if cursor.tables(table="Product", schema="Production").fetchone():
    print("Product table exists")
else:
    print("Product table not found")

检索表的列信息:

# All columns in a table
for row in cursor.columns(table="Product", schema="Production"):
    print(f"{row.column_name}: {row.type_name}({row.column_size})")
    print(f"  Nullable: {row.nullable}, Position: {row.ordinal_position}")

# Filter by column name
for row in cursor.columns(table="Product", schema="Production", column="List%"):
    print(row.column_name)

columns() 参数

用于细化列发现结果的筛选条件:

参数 描述
table 表名称模式。
catalog 目录(数据库)名称。
schema 架构名称模式。
column 列名模式。

columns() 结果列

columns() 方法返回每列的详细信息:

描述
table_cattable_schemtable_name 位置标识符。
column_name 列名称。
data_type SQL 数据类型代码。
type_name 数据类型名称(例如, varcharint)。
column_size 最大长度或精度。
buffer_length 传输缓冲区大小。
decimal_digits 数字类型尺度。
nullable 0 表示非空 1 ,表示可空。
column_def 默认值。
ordinal_position 列位置(从 1 开始)。
is_nullable "YES""NO"

存储过程

探索存储过程:

# List all procedures
for row in cursor.procedures():
    print(f"{row.procedure_schem}.{row.procedure_name}")

# Filter by name pattern
for row in cursor.procedures(procedure="Get%"):
    print(row.procedure_name)

程序()参数

按名称或模式过滤存储过程:

参数 描述
procedure 程序名称模式。
catalog 目录(数据库)名称。
schema 架构名称模式。

procedures() 结果列

该方法返回 procedures() 每个存储过程的元数据:

描述
procedure_catprocedure_schem 位置标识符。
procedure_name 过程名称。
num_input_params 输入参数的数量。
num_output_params 输出参数数量。
num_result_sets 结果集数量。
remarks Description.
procedure_type 类型指示器。

主键

获取表格的主键列:

for row in cursor.primaryKeys(table="Product", schema="Production"):
    print(f"PK column: {row.column_name} (position {row.key_seq})")
    print(f"Constraint name: {row.pk_name}")

primaryKeys() 参数

检索主键信息的参数:

参数 描述
table 表格名称(必填)。
catalog 目录(数据库)名称。
schema 架构名称。

primaryKeys() 结果列

primaryKeys() 方法返回以下信息:

描述
table_cattable_schemtable_name 位置标识符。
column_name 主键中的列。
key_seq 多列键中的位置(从 1 开始计数)。
pk_name 主键约束名称。

外键

发现外国关键关系:

# Foreign keys from a table (outbound references)
for row in cursor.foreignKeys(table="SalesOrderDetail", schema="Sales"):
    print(f"FK {row.fk_name}:")
    print(f"  {row.fktable_name}.{row.fkcolumn_name}")
    print(f"  -> {row.pktable_name}.{row.pkcolumn_name}")

# Foreign keys to a table (inbound references)
for row in cursor.foreignKeys(foreignTable="Product", foreignSchema="Production"):
    print(f"{row.fktable_name} references Product")

foreignKeys() 参数

指定主键表或外键表以发现关系:

参数 描述
table 主键表名称。
catalog 主键目录。
schema 主键模式。
foreignTable 外键表名称。
foreignCatalog 外键目录。
foreignSchema 外键架构。

foreignKeys() 结果列

foreignKeys() 方法返回以下描述关系的列:

描述
pktable_catpktable_schempktable_name 参考(主)表。
pkcolumn_name 参考专栏。
fktable_catfktable_schemfktable_name 引用(外国)表格。
fkcolumn_name 引用栏目。
key_seq 多列键中的位置。
update_rule 对 UPDATE 的操作。
delete_rule 对 DELETE 执行操作。
fk_name 外键约束名称。
pk_name 主键约束名称。

指数与统计

获取表格索引信息:

# All indexes on a table
for row in cursor.statistics(table="Product", schema="Production"):
    if row.index_name:  # Skip table statistics row
        print(f"Index: {row.index_name}")
        print(f"  Column: {row.column_name} (position {row.ordinal_position})")
        print(f"  Unique: {not row.non_unique}")

# Only unique indexes
for row in cursor.statistics(table="Product", schema="Production", unique=True):
    print(f"Unique index: {row.index_name}")

统计() 参数

使用以下过滤器配置索引发现:

参数 默认 描述
table (必需) 表名。
catalog 没有 目录(数据库)名称。
schema 没有 架构名称。
unique 只返回唯一索引。
quick 跳过昂贵的基数/页面检索。

统计()结果列

statistics() 方法返回索引和统计信息:

描述
table_cattable_schemtable_name 位置标识符。
non_unique 0 对于唯一, 1 对于非唯一。
index_name 索引名。
type 索引类型。
ordinal_position 索引中的列位置。
column_name 列名称。
asc_or_desc A 为了上升, D 为了下降。
cardinality 行数估算。
pages 页数。

行标识列

查找可唯一标识某一行的列:

for row in cursor.rowIdColumns(table="Product", schema="Production"):
    print(f"Row ID column: {row.column_name} ({row.type_name})")

该方法返回最佳列集以唯一识别行,可能是主键或唯一索引。

行版本列

查找当任何行值发生变化时自动更新的列。 使用行版本列进行乐观并发控制,你读取行的版本,进行修改,然后确认当前行版本是否相同,再写入:

for row in cursor.rowVerColumns(table="Product", schema="Production"):
    print(f"Version column: {row.column_name}")

结果通常包含用于乐观并发控制的 rowversion/timestamp 列。

数据类型信息

获取支持的SQL数据类型信息:

# All supported types
for row in cursor.getTypeInfo():
    print(f"{row.type_name}: {row.data_type}")
    print(f"  Max size: {row.column_size}")
    print(f"  Nullable: {row.nullable}")

# Specific type
for row in cursor.getTypeInfo(sqlType=mssql_python.SQL_VARCHAR):
    print(f"VARCHAR max size: {row.column_size}")

getTypeInfo() 参数

可选参数用于过滤支持的SQL类型:

参数 描述
sqlType SQL 类型常量(对于所有类型均省略)。

安全注意事项

注意

这些方法会暴露数据库模式的元数据。 虽然方法本身可以安全执行,但返回的信息会揭示你的数据库结构(表名、列名、关系、数据类型)。

  • 不要将原始元数据暴露给不受信任的用户。
  • 在多租户应用程序中清理或筛选结果。
  • 限制对外应用的访问。

示例:生成结构报告

def describe_table(conn, table_name):
    """Generate a schema description for a table."""
    cursor = conn.cursor()
    
    print(f"\n=== {table_name} ===\n")
    
    # Columns
    print("Columns:")
    for col in cursor.columns(table=table_name):
        nullable = "NULL" if col.nullable else "NOT NULL"
        print(f"  {col.column_name}: {col.type_name}({col.column_size}) {nullable}")
    
    # Primary key
    print("\nPrimary Key:")
    pk_cols = cursor.primaryKeys(table=table_name).fetchall()
    if pk_cols:
        pk_names = ", ".join(row.column_name for row in pk_cols)
        print(f"  {pk_cols[0].pk_name}: ({pk_names})")
    else:
        print("  (none)")
    
    # Foreign keys
    print("\nForeign Keys:")
    for fk in cursor.foreignKeys(table=table_name):
        print(f"  {fk.fk_name}: {fk.fkcolumn_name} -> {fk.pktable_name}.{fk.pkcolumn_name}")
    
    # Indexes
    print("\nIndexes:")
    for idx in cursor.statistics(table=table_name):
        if idx.index_name:
            unique = "UNIQUE " if not idx.non_unique else ""
            print(f"  {unique}{idx.index_name}: {idx.column_name}")

# Usage
describe_table(conn, "Product")