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 |
按类型筛选:TABLE, VIEW, SYSTEM TABLE, GLOBAL TEMPORARYLOCAL TEMPORARYALIAS, , 。 SYNONYM |
tables() 结果列
此 tables() 方法为每个表或视图返回以下列:
| 列 | 描述 |
|---|---|
table_cat |
目录(数据库)名称。 |
table_schem |
架构名称。 |
table_name |
表或视图名称。 |
table_type |
TABLE, VIEW, SYSTEM TABLE, GLOBAL TEMPORARY, LOCAL TEMPORARY, ALIAS, , 。 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_cat、table_schem、table_name |
位置标识符。 |
column_name |
列名称。 |
data_type |
SQL 数据类型代码。 |
type_name |
数据类型名称(例如, varchar, int)。 |
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_cat、procedure_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_cat、table_schem、table_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_cat、pktable_schem、pktable_name |
参考(主)表。 |
pkcolumn_name |
参考专栏。 |
fktable_cat、fktable_schem、fktable_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_cat、table_schem、table_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")