Python oracle 테이블 컬럼 비교
DB CAFE
thumb_up 추천메뉴 바로가기
- DBA { Oracle DBA 명령어 > DBA 초급 과정 > DBA 고급 과정 }
- 튜닝 { 오라클 튜닝 목록 }
- 모델링 { 데이터 모델링 가이드 }
1 oracle 테이블 컬럼 비교[편집]
1.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()1.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()1.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()