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() 方法會為每個表格或視圖回傳以下欄位:
| Column | 描述 |
|---|---|
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")
Columns
檢索資料表的欄位資訊:
# 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)
欄位() 參數
用於縮小欄位探索範圍的篩選條件:
| 參數 | 描述 |
|---|---|
table |
資料表名稱模式。 |
catalog |
目錄(資料庫)名稱。 |
schema |
結構名稱模式。 |
column |
欄位名稱格式。 |
columns() 的結果欄位
columns() 方法會回傳每個資料行的詳細資訊:
| Column | 描述 |
|---|---|
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() 回傳每個儲存程序的元資料:
| Column | 描述 |
|---|---|
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() 方法會回傳以下資訊:
| Column | 描述 |
|---|---|
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() 方法回傳以下描述關係的欄位:
| Column | 描述 |
|---|---|
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 |
None | 目錄(資料庫)名稱。 |
schema |
None | 架構名稱。 |
unique |
否 | 只回傳唯一索引。 |
quick |
沒錯 | 跳過昂貴的基數或頁面檢索。 |
統計() 結果欄位
此 statistics() 方法會傳回索引與統計資訊:
| Column | 描述 |
|---|---|
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 型別常數(所有型別皆省略)。 |
安全性考慮
Caution
這些方法會揭露資料庫結構的元資料。 雖然方法本身可以安全執行,但回傳的資訊會揭示你的資料庫結構(資料表名稱、欄位名稱、關聯、資料型別)。
- 不要將原始元資料暴露給不受信任的使用者。
- 在多租戶應用程式中清理或過濾結果。
- 限制對外應用程式的存取權限。
範例:產生結構報告
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")