행위

"Python oracle 테이블 컬럼 비교"의 두 판 사이의 차이

DB CAFE

(새 문서: <source lang=python> import cx_Oracle import pandas as pd from tabulate import tabulate # Oracle DB 연결 설정 def connect_to_oracle(username, password, hostname, port, service_na...)
 
1번째 줄: 1번째 줄:
 +
 +
=== cx_oracle 이용 비교 ===
 +
<source lang=python>
 +
import cx_Oracle
 +
 +
# Oracle DB 연결 설정
 +
def connect_to_oracle(username, password, hostname, port, service_name):
 +
    dsn = cx_Oracle.makedsn(hostname, port, service_name=service_name)
 +
    connection = cx_Oracle.connect(username, password, dsn)
 +
    return connection
 +
 +
# 테이블의 컬럼 정보를 가져오는 함수
 +
def get_table_columns(cursor, table_name):
 +
    query = f"""
 +
    SELECT COLUMN_NAME, DATA_TYPE, DATA_LENGTH, NULLABLE
 +
    FROM ALL_TAB_COLUMNS
 +
    WHERE TABLE_NAME = :table_name
 +
    """
 +
    cursor.execute(query, [table_name.upper()])
 +
    columns = cursor.fetchall()
 +
    return columns
 +
 +
# 두 테이블의 컬럼을 비교하는 함수
 +
def compare_columns(table1_columns, table2_columns):
 +
    table1_columns_set = set(table1_columns)
 +
    table2_columns_set = set(table2_columns)
 +
   
 +
    # 서로 다른 컬럼 찾기
 +
    only_in_table1 = table1_columns_set - table2_columns_set
 +
    only_in_table2 = table2_columns_set - table1_columns_set
 +
   
 +
    return only_in_table1, only_in_table2
 +
 +
# 메인 함수
 +
def main():
 +
    # 데이터베이스 연결 정보
 +
    username = "your_username"
 +
    password = "your_password"
 +
    hostname = "your_hostname"
 +
    port = "1521"
 +
    service_name = "your_service_name"
 +
 +
    # 비교할 테이블 이름
 +
    table1 = "TABLE_A"
 +
    table2 = "TABLE_B"
 +
 +
    # Oracle DB 연결
 +
    connection = connect_to_oracle(username, password, hostname, port, service_name)
 +
    cursor = connection.cursor()
 +
 +
    # 테이블의 컬럼 정보 가져오기
 +
    table1_columns = get_table_columns(cursor, table1)
 +
    table2_columns = get_table_columns(cursor, table2)
 +
 +
    # 컬럼 비교
 +
    only_in_table1, only_in_table2 = compare_columns(table1_columns, table2_columns)
 +
 +
    # 결과 출력
 +
    if only_in_table1:
 +
        print(f"테이블 {table1}에만 존재하는 컬럼: {only_in_table1}")
 +
    else:
 +
        print(f"테이블 {table1}와 {table2}의 모든 컬럼이 동일합니다.")
 +
 +
    if only_in_table2:
 +
        print(f"테이블 {table2}에만 존재하는 컬럼: {only_in_table2}")
 +
    else:
 +
        print(f"테이블 {table2}와 {table1}의 모든 컬럼이 동일합니다.")
 +
 +
    # 연결 종료
 +
    cursor.close()
 +
    connection.close()
 +
 +
if __name__ == "__main__":
 +
    main()
 +
</source>
 +
 +
=== cx_oracle + pandas ===
 +
<source lang=python>
 +
import cx_Oracle
 +
import pandas as pd
 +
 +
# Oracle DB 연결 설정
 +
def connect_to_oracle(username, password, hostname, port, service_name):
 +
    dsn = cx_Oracle.makedsn(hostname, port, service_name=service_name)
 +
    connection = cx_Oracle.connect(username, password, dsn)
 +
    return connection
 +
 +
# 테이블의 컬럼 정보를 가져오는 함수
 +
def get_table_columns_df(cursor, table_name):
 +
    query = f"""
 +
    SELECT COLUMN_NAME, DATA_TYPE, DATA_LENGTH, NULLABLE
 +
    FROM ALL_TAB_COLUMNS
 +
    WHERE TABLE_NAME = :table_name
 +
    """
 +
    cursor.execute(query, [table_name.upper()])
 +
    columns = cursor.fetchall()
 +
   
 +
    # pandas DataFrame으로 변환
 +
    df = pd.DataFrame(columns, columns=["COLUMN_NAME", "DATA_TYPE", "DATA_LENGTH", "NULLABLE"])
 +
    return df
 +
 +
# 두 테이블의 컬럼 정보를 비교하는 함수
 +
def compare_columns_df(df1, df2, table1_name, table2_name):
 +
    # 컬럼 이름으로 비교
 +
    df1_cols = df1[['COLUMN_NAME', 'DATA_TYPE', 'DATA_LENGTH', 'NULLABLE']]
 +
    df2_cols = df2[['COLUMN_NAME', 'DATA_TYPE', 'DATA_LENGTH', 'NULLABLE']]
 +
   
 +
    # 두 DataFrame에서 서로 다른 컬럼 찾기
 +
    diff_table1 = df1_cols[~df1_cols.isin(df2_cols.to_dict(orient="list")).all(axis=1)]
 +
    diff_table2 = df2_cols[~df2_cols.isin(df1_cols.to_dict(orient="list")).all(axis=1)]
 +
 +
    # 결과 출력
 +
    if not diff_table1.empty:
 +
        print(f"\n테이블 {table1_name}에만 존재하는 컬럼들:")
 +
        print(diff_table1)
 +
    else:
 +
        print(f"\n테이블 {table1_name}와 {table2_name}의 모든 컬럼이 동일합니다.")
 +
   
 +
    if not diff_table2.empty:
 +
        print(f"\n테이블 {table2_name}에만 존재하는 컬럼들:")
 +
        print(diff_table2)
 +
    else:
 +
        print(f"\n테이블 {table2_name}와 {table1_name}의 모든 컬럼이 동일합니다.")
 +
 +
# 메인 함수
 +
def main():
 +
    # 데이터베이스 연결 정보
 +
    username = "your_username"
 +
    password = "your_password"
 +
    hostname = "your_hostname"
 +
    port = "1521"
 +
    service_name = "your_service_name"
 +
 +
    # 비교할 테이블 이름
 +
    table1 = "TABLE_A"
 +
    table2 = "TABLE_B"
 +
 +
    # Oracle DB 연결
 +
    connection = connect_to_oracle(username, password, hostname, port, service_name)
 +
    cursor = connection.cursor()
 +
 +
    # 테이블의 컬럼 정보 가져오기
 +
    df1 = get_table_columns_df(cursor, table1)
 +
    df2 = get_table_columns_df(cursor, table2)
 +
 +
    # 컬럼 비교
 +
    compare_columns_df(df1, df2, table1, table2)
 +
 +
    # 연결 종료
 +
    cursor.close()
 +
    connection.close()
 +
 +
if __name__ == "__main__":
 +
    main()
 +
</source>
 +
 +
=== 비교 결과 그리드로 출력 ===
 
<source lang=python>
 
<source lang=python>
 
import cx_Oracle
 
import cx_Oracle

2024년 8월 21일 (수) 18:27 판

thumb_up 추천메뉴 바로가기

1 cx_oracle 이용 비교[편집]

import cx_Oracle

# Oracle DB 연결 설정
def connect_to_oracle(username, password, hostname, port, service_name):
    dsn = cx_Oracle.makedsn(hostname, port, service_name=service_name)
    connection = cx_Oracle.connect(username, password, dsn)
    return connection

# 테이블의 컬럼 정보를 가져오는 함수
def get_table_columns(cursor, table_name):
    query = f"""
    SELECT COLUMN_NAME, DATA_TYPE, DATA_LENGTH, NULLABLE
    FROM ALL_TAB_COLUMNS
    WHERE TABLE_NAME = :table_name
    """
    cursor.execute(query, [table_name.upper()])
    columns = cursor.fetchall()
    return columns

# 두 테이블의 컬럼을 비교하는 함수
def compare_columns(table1_columns, table2_columns):
    table1_columns_set = set(table1_columns)
    table2_columns_set = set(table2_columns)
    
    # 서로 다른 컬럼 찾기
    only_in_table1 = table1_columns_set - table2_columns_set
    only_in_table2 = table2_columns_set - table1_columns_set
    
    return only_in_table1, only_in_table2

# 메인 함수
def main():
    # 데이터베이스 연결 정보
    username = "your_username"
    password = "your_password"
    hostname = "your_hostname"
    port = "1521"
    service_name = "your_service_name"

    # 비교할 테이블 이름
    table1 = "TABLE_A"
    table2 = "TABLE_B"

    # Oracle DB 연결
    connection = connect_to_oracle(username, password, hostname, port, service_name)
    cursor = connection.cursor()

    # 테이블의 컬럼 정보 가져오기
    table1_columns = get_table_columns(cursor, table1)
    table2_columns = get_table_columns(cursor, table2)

    # 컬럼 비교
    only_in_table1, only_in_table2 = compare_columns(table1_columns, table2_columns)

    # 결과 출력
    if only_in_table1:
        print(f"테이블 {table1}에만 존재하는 컬럼: {only_in_table1}")
    else:
        print(f"테이블 {table1}와 {table2}의 모든 컬럼이 동일합니다.")

    if only_in_table2:
        print(f"테이블 {table2}에만 존재하는 컬럼: {only_in_table2}")
    else:
        print(f"테이블 {table2}와 {table1}의 모든 컬럼이 동일합니다.")

    # 연결 종료
    cursor.close()
    connection.close()

if __name__ == "__main__":
    main()

2 cx_oracle + pandas[편집]

import cx_Oracle
import pandas as pd

# Oracle DB 연결 설정
def connect_to_oracle(username, password, hostname, port, service_name):
    dsn = cx_Oracle.makedsn(hostname, port, service_name=service_name)
    connection = cx_Oracle.connect(username, password, dsn)
    return connection

# 테이블의 컬럼 정보를 가져오는 함수
def get_table_columns_df(cursor, table_name):
    query = f"""
    SELECT COLUMN_NAME, DATA_TYPE, DATA_LENGTH, NULLABLE
    FROM ALL_TAB_COLUMNS
    WHERE TABLE_NAME = :table_name
    """
    cursor.execute(query, [table_name.upper()])
    columns = cursor.fetchall()
    
    # pandas DataFrame으로 변환
    df = pd.DataFrame(columns, columns=["COLUMN_NAME", "DATA_TYPE", "DATA_LENGTH", "NULLABLE"])
    return df

# 두 테이블의 컬럼 정보를 비교하는 함수
def compare_columns_df(df1, df2, table1_name, table2_name):
    # 컬럼 이름으로 비교
    df1_cols = df1[['COLUMN_NAME', 'DATA_TYPE', 'DATA_LENGTH', 'NULLABLE']]
    df2_cols = df2[['COLUMN_NAME', 'DATA_TYPE', 'DATA_LENGTH', 'NULLABLE']]
    
    # 두 DataFrame에서 서로 다른 컬럼 찾기
    diff_table1 = df1_cols[~df1_cols.isin(df2_cols.to_dict(orient="list")).all(axis=1)]
    diff_table2 = df2_cols[~df2_cols.isin(df1_cols.to_dict(orient="list")).all(axis=1)]

    # 결과 출력
    if not diff_table1.empty:
        print(f"\n테이블 {table1_name}에만 존재하는 컬럼들:")
        print(diff_table1)
    else:
        print(f"\n테이블 {table1_name}와 {table2_name}의 모든 컬럼이 동일합니다.")
    
    if not diff_table2.empty:
        print(f"\n테이블 {table2_name}에만 존재하는 컬럼들:")
        print(diff_table2)
    else:
        print(f"\n테이블 {table2_name}와 {table1_name}의 모든 컬럼이 동일합니다.")

# 메인 함수
def main():
    # 데이터베이스 연결 정보
    username = "your_username"
    password = "your_password"
    hostname = "your_hostname"
    port = "1521"
    service_name = "your_service_name"

    # 비교할 테이블 이름
    table1 = "TABLE_A"
    table2 = "TABLE_B"

    # Oracle DB 연결
    connection = connect_to_oracle(username, password, hostname, port, service_name)
    cursor = connection.cursor()

    # 테이블의 컬럼 정보 가져오기
    df1 = get_table_columns_df(cursor, table1)
    df2 = get_table_columns_df(cursor, table2)

    # 컬럼 비교
    compare_columns_df(df1, df2, table1, table2)

    # 연결 종료
    cursor.close()
    connection.close()

if __name__ == "__main__":
    main()

3 비교 결과 그리드로 출력[편집]

import cx_Oracle
import pandas as pd
from tabulate import tabulate

# Oracle DB 연결 설정
def connect_to_oracle(username, password, hostname, port, service_name):
    dsn = cx_Oracle.makedsn(hostname, port, service_name=service_name)
    connection = cx_Oracle.connect(username, password, dsn)
    return connection

# 테이블의 컬럼 정보를 가져오는 함수
def get_table_columns_df(cursor, table_name):
    query = f"""
    SELECT COLUMN_NAME, DATA_TYPE, DATA_LENGTH, NULLABLE
    FROM ALL_TAB_COLUMNS
    WHERE TABLE_NAME = :table_name
    """
    cursor.execute(query, [table_name.upper()])
    columns = cursor.fetchall()
    
    # pandas DataFrame으로 변환
    df = pd.DataFrame(columns, columns=["COLUMN_NAME", "DATA_TYPE", "DATA_LENGTH", "NULLABLE"])
    return df

# 두 테이블의 컬럼 정보를 비교하는 함수
def compare_columns_df(df1, df2, table1_name, table2_name):
    # 컬럼 이름으로 비교
    df1_cols = df1[['COLUMN_NAME', 'DATA_TYPE', 'DATA_LENGTH', 'NULLABLE']]
    df2_cols = df2[['COLUMN_NAME', 'DATA_TYPE', 'DATA_LENGTH', 'NULLABLE']]
    
    # 두 DataFrame에서 서로 다른 컬럼 찾기
    diff_table1 = df1_cols[~df1_cols.isin(df2_cols.to_dict(orient="list")).all(axis=1)]
    diff_table2 = df2_cols[~df2_cols.isin(df1_cols.to_dict(orient="list")).all(axis=1)]

    # 그리드 출력
    if not diff_table1.empty:
        print(f"\n테이블 {table1_name}에만 존재하는 컬럼들:")
        print(tabulate(diff_table1, headers='keys', tablefmt='grid'))
    else:
        print(f"\n테이블 {table1_name}와 {table2_name}의 모든 컬럼이 동일합니다.")
    
    if not diff_table2.empty:
        print(f"\n테이블 {table2_name}에만 존재하는 컬럼들:")
        print(tabulate(diff_table2, headers='keys', tablefmt='grid'))
    else:
        print(f"\n테이블 {table2_name}와 {table1_name}의 모든 컬럼이 동일합니다.")

# 메인 함수
def main():
    # 데이터베이스 연결 정보
    username = "your_username"
    password = "your_password"
    hostname = "your_hostname"
    port = "1521"
    service_name = "your_service_name"

    # 비교할 테이블 이름
    table1 = "TABLE_A"
    table2 = "TABLE_B"

    # Oracle DB 연결
    connection = connect_to_oracle(username, password, hostname, port, service_name)
    cursor = connection.cursor()

    # 테이블의 컬럼 정보 가져오기
    df1 = get_table_columns_df(cursor, table1)
    df2 = get_table_columns_df(cursor, table2)

    # 컬럼 비교 및 그리드 출력
    compare_columns_df(df1, df2, table1, table2)

    # 연결 종료
    cursor.close()
    connection.close()

if __name__ == "__main__":
    main()
원본 주소 "http://dbcafe.co.kr/wiki/index.php?title=Python_oracle_테이블_컬럼_비교&oldid=12529"