使用 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() 方法會為每個表格或視圖回傳以下欄位:

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_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() 回傳每個儲存程序的元資料:

Column 描述
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() 方法會回傳以下資訊:

Column 描述
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() 方法回傳以下描述關係的欄位:

Column 描述
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 None 目錄(資料庫)名稱。
schema None 架構名稱。
unique 只回傳唯一索引。
quick 沒錯 跳過昂貴的基數或頁面檢索。

統計() 結果欄位

statistics() 方法會傳回索引與統計資訊:

Column 描述
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 型別常數(所有型別皆省略)。

安全性考慮

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")