mssql-python을 이용한 스키마 탐색

mssql-python 커서 클래스는 ODBC 카탈로그 함수에 매핑되는 9개의 메타데이터 메서드를 제공합니다. 이 방법들을 사용하여 테이블, 열, 저장 프로시저, 키, 인덱스를 프로그래밍적으로 발견할 수 있습니다. 이들은 마이그레이션 도구, 코드 생성기, 관리자 대시보드 등 런타임 시 데이터베이스 스키마에 적응하는 데이터 기반 애플리케이션을 구축하는 데 도움을 줍니다.

Method 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 TEMPORARY, LOCAL TEMPORARY, , ALIAS. 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")

칼럼

테이블의 열 정보 조회:

# 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() 각 열에 대한 상세 정보를 반환합니다:

Column 설명
table_cat, table_schem, table_name 위치 식별자.
column_name 열 이름.
data_type SQL 데이터 타입 코드입니다.
type_name 데이터 타입 이름(예: varchar, ). int
column_size 최대 길이 또는 정밀도.
buffer_length 전송을 위한 버퍼 크기.
decimal_digits 숫자 유형에 대한 척도.
nullable 0 NOT NULL의 경우, 1 NULL 허용의 경우.
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}")

statistics() 매개변수

다음 필터로 인덱스 탐색을 구성하세요:

매개 변수 Default 설명
table (필수) 테이블 이름입니다.
catalog None 카탈로그(데이터베이스) 이름.
schema None 스키마 이름입니다.
unique 거짓 고유 인덱스만 반환합니다.
quick True 비싼 카디널리티나 페이지 검색 과정을 건너뛰세요.

통계() 결과 열

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 타입 상수(모든 타입은 생략).

보안 고려 사항

주의

이 방법들은 데이터베이스 스키마 메타데이터를 노출합니다. 메서드 자체는 안전하게 실행할 수 있지만, 반환된 정보는 데이터베이스 구조(테이블 이름, 열명, 관계, 데이터 타입)를 드러냅니다.

  • 신뢰할 수 없는 사용자에게 원시 메타데이터를 노출하지 마세요.
  • 멀티테넌트 애플리케이션에서 결과를 정제하거나 필터링합니다.
  • 외부와 직접 연결된 애플리케이션의 접근을 제한하세요.

예시: 스키마 보고서 생성

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