Schema-ontdekking met mssql-python

De mssql-python cursorklasse biedt negen metadata-methoden die worden gekoppeld aan ODBC-catalogusfuncties. Gebruik deze methoden om tabellen, kolommen, opgeslagen procedures, sleutels en indexen programmatisch te ontdekken. Zij helpen je bij het bouwen van datagedreven applicaties die zich tijdens runtime aanpassen aan het databaseschema, zoals migratietools, codegeneratoren of beheerdersdashboards.

Elke methode geeft een cursor terug die je kunt itereren om de resultaten te bereiken.

Tables

Geef de tabellen en weergaven in de database weer:

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()-parameters

De volgende parameters regelen de tabeldetectie:

kolommen van het resultaat van tables()

De tables() methode geeft de volgende kolommen terug voor elke tabel of weergave:

Controleer of er een tabel bestaat

Controleer of er een tabel bestaat voordat je deze bevraagt:

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

Columns

Kolominformatie ophalen voor tabellen:

# 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() parameters

Filters voor het verfijnen van het ontdekken van kolommen:

columns() resultaatkolommen

De columns() methode geeft gedetailleerde informatie over elke kolom:

opgeslagen procedures

Ontdek opgeslagen procedures:

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

procedures() parameters

Filter opgeslagen procedures op naam of schema:

resultaatkolommen van procedures()

De procedures() methode levert metadata terug voor elke opgeslagen procedure:

Primaire sleutels

De primaire-sleutelkolommen van een tabel ophalen:

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() parameters

Parameters om primaire sleutelinformatie op te halen:

primaryKeys() resultaatkolommen

De primaryKeys() methode geeft de volgende informatie terug:

Vreemde sleutels

Ontdek buitenlandse sleutelrelaties:

# 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()-parameters

Specificeer primaire sleutel- of vreemde sleuteltabellen om relaties te ontdekken:

foreignKeys() resultaatkolommen

De foreignKeys() methode geeft de volgende kolommen terug die relaties beschrijven:

Indexen en statistieken

Krijg indexinformatie voor een tabel:

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

Statistiek()parameters

Configureer indexontdekking met deze filters:

resultaatkolommen van statistics()

De statistics() methode geeft index- en statistiekinformatie terug:

Kolommen met rijidentificatie

Zoek kolommen die een rij uniek identificeren:

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

Deze methode geeft de beste set kolommen terug om een rij uniek te identificeren, wat de primaire sleutel of een unieke index kan zijn.

Kolommen van rijversies

Zoek kolommen die automatisch worden bijgewerkt wanneer een rijwaarde verandert. Gebruik kolommen van de rijversie voor optimistische gelijktijdigheidscontrole, waarbij je de versie van een rij leest, wijzigingen aanbrengt en vervolgens controleert of de huidige rijversie hetzelfde is voordat je schrijft:

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

Het resultaat bevat doorgaans kolommen rowversion/timestamp die worden gebruikt bij optimistische gelijktijdigheidscontrole.

Gegevenstypegegevens

Informatie over ondersteunde SQL-datatypes:

# 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()-parameters

Optionele parameters om ondersteunde SQL-types te filteren:

Beveiligingsoverwegingen

Voorbeeld: Genereer een schemarapport

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

Verwante onderwerpen

• Voer queries uit
• Gegevens ophalen
• Cursorbeheer