파이썬으로 오라클 스토어드 프로시저 실행하기: 다중 IN/OUT 파라미터와 커서 열 타입 출력 (0개 행 포함)

 

 오라클 스토어드 프로시저에서 SYS_REFCURSOR로 반환되는 emp_names가 0개 행을 반환하더라도, 반환되는 열(first_name, last_name, age)의 데이터 타입(예: VARCHAR2, DECIMAL)을 출력하는 방법을 구현하겠습니다. 이를 위해 oracledb의 커서 객체에서 열 메타데이터를 추출하여 각 열의 데이터 타입을 확인합니다. 또한, 다중 IN/OUT 파라미터를 처리하고, 모든 파라미터의 값, 길이, 타입을 출력합니다.

 

---

핵심 아이디어

1. SQL 파일 읽기: 스토어드 프로시저의 SQL 코드가 담긴 파일을 읽어옵니다.
2. 데이터베이스 연결: oracledb로 오라클 데이터베이스에 연결합니다.
3. 스토어드 프로시저 실행: SQL 코드를 실행해 프로시저를 생성합니다.
4. 다중 IN/OUT 처리: 여러 IN 파라미터를 입력하고, OUT 파라미터(SYS_REFCURSOR, NUMBER, VARCHAR2)를 처리합니다.
5. 열 타입 출력: SYS_REFCURSOR의 열 메타데이터를 추출하여 emp_names의 각 열(first_name, last_name, age)의 데이터 타입(예: VARCHAR2, DECIMAL)을 출력합니다. 0개 행 반환 시에도 열 타입을 확인할 수 있도록 설계합니다.
6. 정보 출력: 모든 IN/OUT 파라미터의 값, 길이, 타입, 그리고 emp_names의 열 타입을 출력합니다.

---

예제 시나리오

• 상황: employees 테이블에서 특정 부서(dept_id)와 직원 상태(status, 예: 'ACTIVE' 또는 'INACTIVE')를 기준으로 직원의 first_name(VARCHAR2), last_name(VARCHAR2), age(DECIMAL)를 반환하는 스토어드 프로시저를 작성했습니다. 추가로, 직원 수(emp_count)와 상태 메시지(status_message)를 OUT 파라미터로 반환합니다.
• SQL 파일: get_employee_details.sql 파일에 아래와 같은 스토어드 프로시저가 작성되어 있습니다.

 

-- get_employee_details.sql
CREATE OR REPLACE PROCEDURE get_employee_details (
    dept_id IN NUMBER,
    status IN VARCHAR2,
    emp_names OUT SYS_REFCURSOR,
    emp_count OUT NUMBER,
    status_message OUT VARCHAR2
) AS
BEGIN
    -- 직원 정보 커서 열기
    OPEN emp_names FOR
        SELECT first_name, last_name, age
        FROM employees
        WHERE department_id = dept_id
        AND employee_status = status;

    -- 직원 수 계산
    SELECT COUNT(*)
    INTO emp_count
    FROM employees
    WHERE department_id = dept_id
    AND employee_status = status;

    -- 상태 메시지 설정
    IF emp_count > 0 THEN
        status_message := 'Success: Found ' || emp_count || ' employees';
    ELSE
        status_message := 'No employees found for dept_id ' || dept_id || ' and status ' || status;
    END IF;
END;
/

• 목표:
  • IN 파라미터: dept_id(NUMBER), status(VARCHAR2)를 입력.
  • OUT 파라미터: emp_names(SYS_REFCURSOR, first_name, last_name, age의 리스트), emp_count(NUMBER), status_message(VARCHAR2)를 반환.
  • emp_names의 각 열(first_name, last_name, age)의 데이터 타입(예: VARCHAR2, VARCHAR2, DECIMAL)을 출력. 0개 행 반환 시에도 열 타입을 확인.
  • 각 파라미터의 값, 길이, 타입을 출력.
  • 0개 행 반환 시(예: 부서에 직원이 없거나 상태에 맞는 직원이 없음)도 안정적으로 처리.

---

 

 

 

파이썬 코드 예시

아래 코드는 SQL 파일을 읽어 스토어드 프로시저를 실행하고, 다중 IN/OUT 파라미터를 처리하며, SYS_REFCURSOR의 열 타입(first_name: VARCHAR2, last_name: VARCHAR2, age: DECIMAL)을 출력합니다. 0개 행 반환 시에도 열 타입을 추출할 수 있도록 설계했습니다.

 

 

import oracledb

def execute_stored_procedure_from_file(db_config, file_path):
    """
    SQL 파일에서 스토어드 프로시저 코드를 읽어와 오라클 데이터베이스에서 실행합니다.

    Args:
        db_config (dict): 데이터베이스 연결 정보 (예: {'user': '...', 'password': '...', 'dsn': '...'}).
        file_path (str): 스토어드 프로시저 SQL 코드 파일 경로.
    """
    connection = None
    cursor = None
    try:
        # 데이터베이스 연결
        connection = oracledb.connect(**db_config)
        cursor = connection.cursor()

        # SQL 파일 읽기
        with open(file_path, 'r') as f:
            sql_code = f.read()

        # SQL 실행 (여러 구문 처리)
        for statement in sql_code.split('/'):
            statement = statement.strip()
            if statement:
                cursor.execute(statement)
        connection.commit()
        print(f"'{file_path}'의 스토어드 프로시저가 성공적으로 실행되었습니다.")

    except oracledb.Error as error:
        print(f"오라클 오류: {error}")
        if connection:
            connection.rollback()
    finally:
        if cursor:
            cursor.close()
        if connection:
            connection.close()

def call_get_employee_details(db_config, dept_id, status):
    """
    get_employee_details 프로시저를 호출하여 직원 정보와 추가 정보를 반환하고,
    모든 IN/OUT 파라미터의 값, 길이, 타입 및 emp_names의 열 타입을 출력합니다.

    Args:
        db_config (dict): 데이터베이스 연결 정보.
        dept_id (int): 부서 ID.
        status (str): 직원 상태 (예: 'ACTIVE', 'INACTIVE').
    """
    connection = None
    cursor = None
    try:
        connection = oracledb.connect(**db_config)
        cursor = connection.cursor()

        # OUT 파라미터 변수 정의
        emp_names_cursor = cursor.var(oracledb.CURSOR)
        emp_count = cursor.var(oracledb.NUMBER)
        status_message = cursor.var(oracledb.STRING)

        # 프로시저 호출
        cursor.callproc('get_employee_details', [dept_id, status, emp_names_cursor, emp_count, status_message])

        # 커서에서 결과 가져오기 (다중 열 처리)
        emp_names = [list(row) for row in emp_names_cursor.getvalue()]

        # 열 메타데이터 추출 (열 이름 및 타입)
        cursor_value = emp_names_cursor.getvalue()
        column_types = []
        if cursor_value.description:
            for col in cursor_value.description:
                col_name = col[0]  # 열 이름
                col_type = col[1]  # 열 타입
                # oracledb 타입을 오라클 SQL 타입으로 매핑
                if col_type == oracledb.DB_TYPE_VARCHAR:
                    col_type_name = "VARCHAR2"
                elif col_type == oracledb.DB_TYPE_NUMBER:
                    col_type_name = "DECIMAL" if col[5] > 0 else "NUMBER"  # 소수점이 있으면 DECIMAL
                else:
                    col_type_name = str(col_type)
                column_types.append((col_name, col_type_name))
        else:
            # 0개 행이라도 기본 열 타입을 수동으로 정의 (쿼리 구조 기반)
            column_types = [
                ("FIRST_NAME", "VARCHAR2"),
                ("LAST_NAME", "VARCHAR2"),
                ("AGE", "DECIMAL")
            ]

        # 정보 출력
        print(f"IN 파라미터:")
        print(f"  dept_id: {dept_id}, 타입: {type(dept_id)}")
        print(f"  status: {status}, 타입: {type(status)}")
        print(f"OUT 파라미터:")
        print(f"  emp_names: {emp_names}")
        print(f"  emp_names 길이: {len(emp_names)}")
        print(f"  emp_names 타입: {type(emp_names)}")
        print(f"  emp_names 열 타입: {column_types}")
        print(f"  emp_count: {emp_count.getvalue()}")
        print(f"  emp_count 타입: {type(emp_count.getvalue())}")
        print(f"  status_message: {status_message.getvalue()}")
        print(f"  status_message 타입: {type(status_message.getvalue())}")

        return emp_names, emp_count.getvalue(), status_message.getvalue()

    except oracledb.Error as error:
        print(f"오라클 오류: {error}")
    finally:
        if cursor:
            cursor.close()
        if connection:
            connection.close()

# 데이터베이스 연결 정보 설정
db_config = {
    'user': 'your_username',
    'password': 'your_password',
    'dsn': 'your_hostname:1521/your_service_name'
}

# SQL 파일 경로
sql_file_path = 'path/to/get_employee_details.sql'

# 1. 스토어드 프로시저 실행
execute_stored_procedure_from_file(db_config, sql_file_path)

# 2. 스토어드 프로시저 호출 (부서 ID 10, 상태 'ACTIVE' - 직원 3명 있다고 가정)
print("\n부서 ID 10, 상태 'ACTIVE' 테스트:")
call_get_employee_details(db_config, 10, 'ACTIVE')

# 3. 스토어드 프로시저 호출 (부서 ID 999, 상태 'INACTIVE' - 직원 0명)
print("\n부서 ID 999, 상태 'INACTIVE' 테스트 (0개 행):")
call_get_employee_details(db_config, 999, 'INACTIVE')

---

코드 설명

1. import oracledb: 오라클 데이터베이스 연결을 위한 라이브러리입니다. 설치가 필요하면 pip install oracledb를 실행하세요.
2. execute_stored_procedure_from_file 함수:
   • 파일 읽기: with open(file_path, 'r') as f로 SQL 파일을 읽습니다.
   • 여러 구문 처리: /로 구분된 SQL 구문을 분리해 실행합니다.
   • 오류 처리: try-except-finally 블록으로 오류를 처리하고 리소스를 정리합니다.
3. call_get_employee_details 함수:
   • OUT 파라미터 정의:
     • emp_names_cursor = cursor.var(oracledb.CURSOR): SYS_REFCURSOR로 다중 열 데이터 처리.
     • emp_count = cursor.var(oracledb.NUMBER): NUMBER 타입의 직원 수.
     • status_message = cursor.var(oracledb.STRING): VARCHAR2 타입의 상태 메시지.
   • 프로시저 호출: cursor.callproc로 get_employee_details 프로시저를 호출하며, IN 파라미터(dept_id, status)와 OUT 파라미터를 전달합니다.
   • 다중 열 처리: emp_names_cursor.getvalue()로 커서의 결과를 가져오고, [list(row) for row in ...]로 각 행을 리스트로 변환합니다. 행이 0개면 emp_names는 빈 리스트([])가 됩니다.
   • 열 메타데이터 추출:
     • cursor_value.description을 사용해 열 이름과 타입을 가져옵니다.
     • oracledb.DB_TYPE_VARCHAR는 VARCHAR2로, oracledb.DB_TYPE_NUMBER는 소수점이 있으면 DECIMAL, 없으면 NUMBER로 매핑합니다.
     • 0개 행일 경우 description이 None일 수 있으므로, 쿼리 구조를 기반으로 기본 열 타입(FIRST_NAME: VARCHAR2, LAST_NAME: VARCHAR2, AGE: DECIMAL)을 수동으로 정의합니다.
   • 정보 출력:
     • IN 파라미터: dept_id, status의 값과 타입.
     • OUT 파라미터: emp_names(리스트), emp_count(숫자), status_message(문자열)의 값, 길이(리스트의 경우), 타입.
     • emp_names 열 타입: 각 열의 이름과 오라클 데이터 타입(예: [(FIRST_NAME, VARCHAR2), (LAST_NAME, VARCHAR2), (AGE, DECIMAL)]).
   • 반환: emp_names, emp_count, status_message를 반환합니다.
4. 연결 정보: db_config에 실제 데이터베이스 정보(사용자 이름, 비밀번호, DSN)를 입력하세요.
5. 실행 흐름:
   • execute_stored_procedure_from_file로 SQL 파일을 실행해 프로시저를 생성합니다.
   • call_get_employee_details를 두 번 호출해:
     • 부서 ID 10, 상태 'ACTIVE'(직원 3명 가정).
     • 부서 ID 999, 상태 'INACTIVE'(직원 0명 가정)을 테스트합니다.

---

 

 

실행 결과 예시

get_employee_details.sql이 성공적으로 실행되고, 아래와 같은 상황을 가정합니다:

• 부서 ID 10, 상태 'ACTIVE': 직원 3명(John Doe, 30세; Jane Smith, 25세; Bob Johnson, 40세).
• 부서 ID 999, 상태 'INACTIVE': 직원 0명.

출력은 다음과 같습니다:

 

'path/to/get_employee_details.sql'의 스토어드 프로시저가 성공적으로 실행되었습니다.

부서 ID 10, 상태 'ACTIVE' 테스트:
IN 파라미터:
  dept_id: 10, 타입: <class 'int'>
  status: ACTIVE, 타입: <class 'str'>
OUT 파라미터:
  emp_names: [['John', 'Doe', 30], ['Jane', 'Smith', 25], ['Bob', 'Johnson', 40]]
  emp_names 길이: 3
  emp_names 타입: <class 'list'>
  emp_names 열 타입: [('FIRST_NAME', 'VARCHAR2'), ('LAST_NAME', 'VARCHAR2'), ('AGE', 'DECIMAL')]
  emp_count: 3
  emp_count 타입: <class 'float'>
  status_message: Success: Found 3 employees
  status_message 타입: <class 'str'>

부서 ID 999, 상태 'INACTIVE' 테스트 (0개 행):
IN 파라미터:
  dept_id: 999, 타입: <class 'int'>
  status: INACTIVE, 타입: <class 'str'>
OUT 파라미터:
  emp_names: []
  emp_names 길이: 0
  emp_names 타입: <class 'list'>
  emp_names 열 타입: [('FIRST_NAME', 'VARCHAR2'), ('LAST_NAME', 'VARCHAR2'), ('AGE', 'DECIMAL')]
  emp_count: 0
  emp_count 타입: <class 'float'>
  status_message: No employees found for dept_id 999 and status INACTIVE
  status_message 타입: <class 'str'>

---

추가 설명: 열 타입 추출의 세부 사항

• 정상적인 경우 (행이 1개 이상): cursor_value.description은 열 이름, 타입, 크기 등의 메타데이터를 제공합니다. 이를 통해 first_name(VARCHAR2), last_name(VARCHAR2), age(DECIMAL)를 정확히 매핑합니다.
• 0개 행인 경우: description이 None일 수 있으므로, 쿼리 구조를 기반으로 기본 열 타입을 수동으로 정의했습니다. 실제 환경에서는 스토어드 프로시저의 쿼리(SELECT first_name, last_name, age)를 분석하거나, 데이터베이스 스키마를 참조하여 열 타입을 미리 알아야 합니다.
• 타입 매핑: oracledb.DB_TYPE_NUMBER는 소수점 여부(col[5] > 0)에 따라 DECIMAL 또는 NUMBER로 구분했습니다. 필요에 따라 더 정밀한 매핑(예: INTEGER, FLOAT)을 추가할 수 있습니다.

---

주의사항

1. SQL 파일 형식: SQL 파일은 올바른 DDL 구문(CREATE OR REPLACE PROCEDURE)을 포함해야 하며, /로 구분된 구문이 있다면 코드에서 이를 처리합니다.
2. 데이터베이스 환경: employees 테이블과 first_name, last_name, age, department_id, employee_status 열이 실제로 존재해야 합니다. 예제에 맞는 테이블이 없다면 자신의 데이터베이스에 맞게 수정하세요.
3. 보안: 연결 정보를 코드에 직접 포함하는 것은 보안 위험이 있습니다. 환경 변수나 설정 파일을 사용하세요.
4. 오류 처리: 운영 환경에서는 에러 로그를 파일에 저장하거나, 더 구체적인 예외 처리를 추가하세요.
5. 열 타입 수동 정의: 0개 행일 때 열 타입을 수동으로 정의했으므로, 쿼리 구조가 변경되면 column_types 리스트를 업데이트해야 합니다.
6. NUMBER 타입: emp_count와 age는 NUMBER로 정의되었지만, oracledb에서 float으로 반환될 수 있습니다. 필요하면 int()로 변환하세요.

---

마무리

이 예제를 통해 oracledb로 오라클 스토어드 프로시저에서 다중 IN 파라미터(dept_id, status)와 다중 OUT 파라미터(emp_names, emp_count, status_message)를 처리하고, emp_names의 열 타입(first_name: VARCHAR2, last_name: VARCHAR2, age: DECIMAL)을 출력하는 방법을 배웠습니다. 특히, 0개 행 반환 시에도 열 타입을 안정적으로 출력하도록 설계했습니다.

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

---

 

0개 행일 때 SYS_REFCURSOR (emp_names)의 열 메타데이터(cursor.description)를 동적으로 알아내고, 다양한 열 타입(예: VARCHAR2, NUMBER, DATE, CHAR, CLOB)을 포함한 emp_names를 처리하도록 코드를 수정하겠습니다. 또한, emp_names가 0건일 때 열 메타데이터를 기반으로 JSON 데이터 10건을 생성하며, 여러 시나리오(다양한 필드 타입과 구조)를 예제로 설명하겠습니다. 이를 위해 별도의 SELECT ... WHERE 1=0 쿼리를 사용해 메타데이터를 추출하는 방법을 유지합니다.

---

목표

• 스토어드 프로시저: get_employee_details에서 emp_names (SYS_REFCURSOR) 반환.
• 열 타입 다양화: VARCHAR2, NUMBER, DATE, CHAR, CLOB 등 다양한 타입 포함.
• 0개 행 처리: cursor.description이 None일 때, 별도 쿼리로 메타데이터 추출.
• JSON 생성: emp_names가 0건일 때, 열 메타데이터를 기반으로 10건의 가상 JSON 데이터 생성.
• 시나리오: 다양한 필드 조합(예: 기본 정보, 상세 정보, 상태 정보)으로 테스트.

---

스토어드 프로시저 (가정)

get_employee_details.sql은 아래와 같다고 가정합니다. 각 시나리오에 따라 SELECT 문의 열 구조가 다릅니다

-- get_employee_details.sql
CREATE OR REPLACE PROCEDURE get_employee_details (
    dept_id IN NUMBER,
    status IN VARCHAR2,
    emp_names OUT SYS_REFCURSOR,
    emp_count OUT NUMBER,
    status_message OUT VARCHAR2
) AS
BEGIN
    -- 시나리오에 따라 SELECT 문 변경 (아래는 기본 시나리오)
    OPEN emp_names FOR
        SELECT first_name, last_name, hire_date, salary, status_code
        FROM employees
        WHERE department_id = dept_id
        AND employee_status = status;

    SELECT COUNT(*)
    INTO emp_count
    FROM employees
    WHERE department_id = dept_id
    AND employee_status = status;

    IF emp_count > 0 THEN
        status_message := 'Success: Found ' || emp_count || ' employees';
    ELSE
        status_message := 'No employees found for dept_id ' || dept_id || ' and status ' || status;
    END IF;
END;
/

---

시나리오 정의

다양한 필드 타입을 테스트하기 위해 세 가지 시나리오를 정의합니다. 각 시나리오는 emp_names의 열 구조가 다르며, 이를 동적으로 처리합니다.

1. 시나리오 1: 기본 직원 정보
   • 열:
     • first_name: VARCHAR2 (이름, 예: "John").
     • last_name: VARCHAR2 (성, 예: "Doe").
     • hire_date: DATE (입사일, 예: "2020-06-15").
     • salary: NUMBER (급여, 예: 50000).
     • status_code: CHAR(1) (상태, 예: "A" for Active).
   • 쿼리: SELECT first_name, last_name, hire_date, salary, status_code FROM employees WHERE ...
2. 시나리오 2: 상세 직원 정보
   • 열:
     • employee_id: NUMBER (사원 ID, 예: 1001).
     • email: VARCHAR2 (이메일, 예: "john.doe@example.com").
     • birth_date: DATE (생일, 예: "1990-03-22").
     • department_name: VARCHAR2 (부서명, 예: "Engineering").
     • notes: CLOB (메모, 예: "Excellent performer").
   • 쿼리: SELECT employee_id, email, birth_date, department_name, notes FROM employees WHERE ...
3. 시나리오 3: 최소 정보
   • 열:
     • full_name: VARCHAR2 (전체 이름, 예: "John Doe").
     • is_active: CHAR(1) (활성 여부, 예: "Y").
   • 쿼리: SELECT full_name, is_active FROM employees WHERE ...

---

수정된 파이썬 코드

아래 코드는 emp_names의 열 메타데이터를 동적으로 처리하고, 0개 행일 때 SELECT ... WHERE 1=0 쿼리로 메타데이터를 추출합니다. 각 시나리오에 맞는 쿼리를 설정하여 테스트하며, 다양한 열 타입(VARCHAR2, NUMBER, DATE, CHAR, CLOB)에 맞는 가상 JSON 데이터를 생성합니다.

import oracledb
import json
from faker import Faker
import random
from datetime import datetime, timedelta

def get_column_metadata(db_config, query, bind_vars):
    """
    주어진 쿼리를 실행하여 열 메타데이터를 추출합니다 (0개 행 반환).

    Args:
        db_config (dict): 데이터베이스 연결 정보.
        query (str): 메타데이터를 얻기 위한 SELECT 쿼리.
        bind_vars (list): 쿼리의 바인드 변수 값.

    Returns:
        list: 열 이름과 타입 리스트 (예: [('FIRST_NAME', 'VARCHAR2'), ...]).
    """
    connection = None
    cursor = None
    try:
        connection = oracledb.connect(**db_config)
        cursor = connection.cursor()
        cursor.execute(query, bind_vars)
        column_types = []
        for col in cursor.description:
            col_name = col[0]
            col_type = col[1]
            if col_type == oracledb.DB_TYPE_VARCHAR:
                col_type_name = "VARCHAR2"
            elif col_type == oracledb.DB_TYPE_DATE:
                col_type_name = "DATE"
            elif col_type == oracledb.DB_TYPE_NUMBER:
                col_type_name = "DECIMAL" if col[5] > 0 else "NUMBER"
            elif col_type == oracledb.DB_TYPE_CHAR:
                col_type_name = "CHAR"
            elif col_type == oracledb.DB_TYPE_CLOB:
                col_type_name = "CLOB"
            else:
                col_type_name = str(col_type)
            column_types.append((col_name, col_type_name))
        return column_types
    except oracledb.Error as error:
        print(f"메타데이터 추출 오류: {error}")
        return []
    finally:
        if cursor:
            cursor.close()
        if connection:
            connection.close()

def generate_dummy_json_data(column_types, num_records=10):
    """
    열 타입 정보를 기반으로 가상의 JSON 데이터 10건을 동적으로 생성합니다.

    Args:
        column_types (list): 열 이름과 타입 리스트.
        num_records (int): 생성할 레코드 수 (기본값: 10).

    Returns:
        list: JSON 형식으로 변환 가능한 딕셔너리 리스트.
    """
    faker = Faker()
    dummy_data = []

    for _ in range(num_records):
        record = {}
        for col_name, col_type in column_types:
            col_name_lower = col_name.lower()
            if col_type == "VARCHAR2":
                if col_name_lower == "first_name":
                    record[col_name] = faker.first_name()
                elif col_name_lower == "last_name":
                    record[col_name] = faker.last_name()
                elif col_name_lower == "email":
                    record[col_name] = faker.email()
                elif col_name_lower == "department_name":
                    record[col_name] = faker.company()
                elif col_name_lower == "full_name":
                    record[col_name] = faker.name()
                else:
                    record[col_name] = faker.word()[:30]
            elif col_type == "DATE":
                days_offset = random.randint(-3650, 0) if col_name_lower == "birth_date" else random.randint(-30, 30)
                record[col_name] = (datetime.now() + timedelta(days=days_offset)).strftime("%Y-%m-%d")
            elif col_type in ("DECIMAL", "NUMBER"):
                if col_name_lower == "salary":
                    record[col_name] = round(random.uniform(30000, 150000), 2)
                elif col_name_lower == "employee_id":
                    record[col_name] = random.randint(1000, 9999)
                else:
                    record[col_name] = random.randint(1, 100)
            elif col_type == "CHAR":
                record[col_name] = random.choice(["A", "I"]) if col_name_lower == "status_code" else random.choice(["Y", "N"])
            elif col_type == "CLOB":
                record[col_name] = faker.text(max_nb_chars=200)
            else:
                record[col_name] = None
        dummy_data.append(record)

    return dummy_data

def execute_stored_procedure_from_file(db_config, file_path):
    """
    SQL 파일에서 스토어드 프로시저 코드를 읽어와 실행합니다.
    """
    connection = None
    cursor = None
    try:
        connection = oracledb.connect(**db_config)
        cursor = connection.cursor()
        with open(file_path, 'r') as f:
            sql_code = f.read()
        for statement in sql_code.split('/'):
            statement = statement.strip()
            if statement:
                cursor.execute(statement)
        connection.commit()
        print(f"'{file_path}'의 스토어드 프로시저가 성공적으로 실행되었습니다.")
    except oracledb.Error as error:
        print(f"오라클 오류: {error}")
        if connection:
            connection.rollback()
    finally:
        if cursor:
            cursor.close()
        if connection:
            connection.close()

def call_get_employee_details(db_config, dept_id, status, metadata_query):
    """
    get_employee_details 프로시저를 호출하고, 열 메타데이터를 동적으로 처리합니다.
    0개 행일 경우 별도 쿼리로 메타데이터를 추출하여 JSON 데이터를 생성합니다.

    Args:
        db_config (dict): 데이터베이스 연결 정보.
        dept_id (int): 부서 ID.
        status (str): 직원 상태 (예: 'ACTIVE', 'INACTIVE').
        metadata_query (str): 0개 행일 때 메타데이터를 얻기 위한 쿼리.

    Returns:
        tuple: (emp_names 또는 dummy_json_data, emp_count, status_message).
    """
    connection = None
    cursor = None
    try:
        connection = oracledb.connect(**db_config)
        cursor = connection.cursor()

        # OUT 파라미터 정의
        emp_names_cursor = cursor.var(oracledb.CURSOR)
        emp_count = cursor.var(oracledb.NUMBER)
        status_message = cursor.var(oracledb.STRING)

        # 프로시저 호출
        cursor.callproc('get_employee_details', [dept_id, status, emp_names_cursor, emp_count, status_message])

        # 커서에서 결과 가져오기
        emp_names = [list(row) for row in emp_names_cursor.getvalue()]

        # 열 메타데이터 추출
        cursor_value = emp_names_cursor.getvalue()
        column_types = []
        if cursor_value.description:
            for col in cursor_value.description:
                col_name = col[0]
                col_type = col[1]
                if col_type == oracledb.DB_TYPE_VARCHAR:
                    col_type_name = "VARCHAR2"
                elif col_type == oracledb.DB_TYPE_DATE:
                    col_type_name = "DATE"
                elif col_type == oracledb.DB_TYPE_NUMBER:
                    col_type_name = "DECIMAL" if col[5] > 0 else "NUMBER"
                elif col_type == oracledb.DB_TYPE_CHAR:
                    col_type_name = "CHAR"
                elif col_type == oracledb.DB_TYPE_CLOB:
                    col_type_name = "CLOB"
                else:
                    col_type_name = str(col_type)
                column_types.append((col_name, col_type_name))
        else:
            # 0개 행일 경우 별도 쿼리로 메타데이터 추출
            column_types = get_column_metadata(db_config, metadata_query, [])
            if not column_types:
                print("Error: Could not retrieve column metadata.")
                return [], None, None

        # emp_names 길이가 0이면 가상의 JSON 데이터 생성
        if len(emp_names) == 0:
            print("emp_names가 비어있으므로 가상의 JSON 데이터 10건을 생성합니다.")
            dummy_json_data = generate_dummy_json_data(column_types)
        else:
            dummy_json_data = [
                {column_types[i][0]: value for i, value in enumerate(row)}
                for row in emp_names
            ]

        # 정보 출력
        print(f"IN 파라미터:")
        print(f"  dept_id: {dept_id}, 타입: {type(dept_id)}")
        print(f"  status: {status}, 타입: {type(status)}")
        print(f"OUT 파라미터:")
        print(f"  emp_names: {emp_names}")
        print(f"  emp_names 길이: {len(emp_names)}")
        print(f"  emp_names 타입: {type(emp_names)}")
        print(f"  emp_names 열 타입: {column_types}")
        print(f"  emp_count: {emp_count.getvalue()}")
        print(f"  emp_count 타입: {type(emp_count.getvalue())}")
        print(f"  status_message: {status_message.getvalue()}")
        print(f"  status_message 타입: {type(status_message.getvalue())}")
        print(f"JSON 데이터: {json.dumps(dummy_json_data, indent=2)}")

        return dummy_json_data, emp_count.getvalue(), status_message.getvalue()

    except oracledb.Error as error:
        print(f"오라클 오류: {error}")
        return [], None, None
    finally:
        if cursor:
            cursor.close()
        if connection:
            connection.close()

# 데이터베이스 연결 정보 설정
db_config = {
    'user': 'your_username',
    'password': 'your_password',
    'dsn': 'your_hostname:1521/your_service_name'
}

# SQL 파일 경로
sql_file_path = 'path/to/get_employee_details.sql'

# 시나리오별 메타데이터 쿼리
scenario_queries = {
    "scenario1": """
        SELECT first_name, last_name, hire_date, salary, status_code
        FROM employees
        WHERE 1=0
    """,
    "scenario2": """
        SELECT employee_id, email, birth_date, department_name, notes
        FROM employees
        WHERE 1=0
    """,
    "scenario3": """
        SELECT full_name, is_active
        FROM employees
        WHERE 1=0
    """
}

# 1. 스토어드 프로시저 실행
execute_stored_procedure_from_file(db_config, sql_file_path)

# 2. 시나리오별 테스트
# 시나리오 1: 기본 직원 정보 (dept_id=10, status='ACTIVE', 3건 가정)
print("\n시나리오 1: 기본 직원 정보 (dept_id=10, status='ACTIVE')")
call_get_employee_details(db_config, 10, 'ACTIVE', scenario_queries["scenario1"])

# 시나리오 1: 0개 행 (dept_id=999, status='INACTIVE')
print("\n시나리오 1: 기본 직원 정보 (dept_id=999, status='INACTIVE', 0개 행)")
call_get_employee_details(db_config, 999, 'INACTIVE', scenario_queries["scenario1"])

# 시나리오 2: 상세 직원 정보 (dept_id=10, status='ACTIVE', 0개 행 가정)
print("\n시나리오 2: 상세 직원 정보 (dept_id=10, status='ACTIVE', 0개 행)")
call_get_employee_details(db_config, 10, 'ACTIVE', scenario_queries["scenario2"])

# 시나리오 3: 최소 정보 (dept_id=10, status='ACTIVE', 0개 행 가정)
print("\n시나리오 3: 최소 정보 (dept_id=10, status='ACTIVE', 0개 행)")
call_get_employee_details(db_config, 10, 'ACTIVE', scenario_queries["scenario3"])

---

코드 설명

1. get_column_metadata:
   • 목적: SELECT ... WHERE 1=0 쿼리를 실행하여 열 메타데이터 추출.
   • 타입 매핑:
     • DB_TYPE_VARCHAR → VARCHAR2.
     • DB_TYPE_DATE → DATE.
     • DB_TYPE_NUMBER → DECIMAL (소수점 있음) 또는 NUMBER.
     • DB_TYPE_CHAR → CHAR.
     • DB_TYPE_CLOB → CLOB.
   • 반환: [(col_name, col_type_name), ...].
2. generate_dummy_json_data:
   • 동적 데이터 생성:
     • VARCHAR2:
       • first_name, last_name, full_name: faker로 이름 생성.
       • email: faker.email().
       • department_name: faker.company().
       • 기타: faker.word().
     • DATE:
       • birth_date: 10년 이내 과거 날짜.
       • hire_date: ±30일 내 날짜.
     • NUMBER/DECIMAL:
       • employee_id: 1000~9999.
       • salary: 30000~150000, 소수점 2자리.
       • 기타: 1~100.
     • CHAR:
       • status_code: A 또는 I.
       • is_active: Y 또는 N.
     • CLOB: faker.text()로 200자 이내 텍스트.
   • 출력: JSON 딕셔너리 리스트.
3. call_get_employee_details:
   • 0개 행 처리:
     • cursor.description이 None이면 get_column_metadata로 메타데이터 추출.
     • metadata_query는 시나리오별로 제공.
   • JSON 생성:
     • emp_names가 비어있으면 generate_dummy_json_data로 10건 생성.
     • 비어있지 않으면 emp_names를 JSON으로 변환.
   • 출력: IN/OUT 파라미터, 열 타입, JSON 데이터.
4. 시나리오 테스트:
   • 각 시나리오에 맞는 metadata_query를 scenario_queries 딕셔너리로 관리.
   • 실제 프로시저의 SELECT 문과 metadata_query가 동일해야 함.

---

실행 결과 예시

시나리오 1: 기본 직원 정보 (dept_id=10, status='ACTIVE', 3건 가정)

 

시나리오 1: 기본 직원 정보 (dept_id=10, status='ACTIVE')
IN 파라미터:
  dept_id: 10, 타입: <class 'int'>
  status: ACTIVE, 타입: <class 'str'>
OUT 파라미터:
  emp_names: [['John', 'Doe', '2020-06-15', 50000, 'A'], ['Jane', 'Smith', '2021-03-10', 60000, 'A'], ['Bob', 'Johnson', '2019-11-22', 75000, 'A']]
  emp_names 길이: 3
  emp_names 타입: <class 'list'>
  emp_names 열 타입: [('FIRST_NAME', 'VARCHAR2'), ('LAST_NAME', 'VARCHAR2'), ('HIRE_DATE', 'DATE'), ('SALARY', 'NUMBER'), ('STATUS_CODE', 'CHAR')]
  emp_count: 3
  emp_count 타입: <class 'float'>
  status_message: Success: Found 3 employees
  status_message 타입: <class 'str'>
JSON 데이터: [
  {
    "FIRST_NAME": "John",
    "LAST_NAME": "Doe",
    "HIRE_DATE": "2020-06-15",
    "SALARY": 50000,
    "STATUS_CODE": "A"
  },
  {
    "FIRST_NAME": "Jane",
    "LAST_NAME": "Smith",
    "HIRE_DATE": "2021-03-10",
    "SALARY": 60000,
    "STATUS_CODE": "A"
  },
  {
    "FIRST_NAME": "Bob",
    "LAST_NAME": "Johnson",
    "HIRE_DATE": "2019-11-22",
    "SALARY": 75000,
    "STATUS_CODE": "A"
  }
]

시나리오 1: 기본 직원 정보 (dept_id=999, status='INACTIVE', 0건)

 

시나리오 1: 기본 직원 정보 (dept_id=999, status='INACTIVE', 0개 행)
emp_names가 비어있으므로 가상의 JSON 데이터 10건을 생성합니다.
IN 파라미터:
  dept_id: 999, 타입: <class 'int'>
  status: INACTIVE, 타입: <class 'str'>
OUT 파라미터:
  emp_names: []
  emp_names 길이: 0
  emp_names 타입: <class 'list'>
  emp_names 열 타입: [('FIRST_NAME', 'VARCHAR2'), ('LAST_NAME', 'VARCHAR2'), ('HIRE_DATE', 'DATE'), ('SALARY', 'NUMBER'), ('STATUS_CODE', 'CHAR')]
  emp_count: 0
  emp_count 타입: <class 'float'>
  status_message: No employees found for dept_id 999 and status INACTIVE
  status_message 타입: <class 'str'>
JSON 데이터: [
  {
    "FIRST_NAME": "Michael",
    "LAST_NAME": "Brown",
    "HIRE_DATE": "2025-04-10",
    "SALARY": 82000.45,
    "STATUS_CODE": "A"
  },
  ...
  {
    "FIRST_NAME": "Lisa",
    "LAST_NAME": "Martin",
    "HIRE_DATE": "2025-05-01",
    "SALARY": 95000.12,
    "STATUS_CODE": "I"
  }
]

시나리오 2: 상세 직원 정보 (dept_id=10, status='ACTIVE', 0건)

 

시나리오 2: 상세 직원 정보 (dept_id=10, status='ACTIVE', 0개 행)
emp_names가 비어있으므로 가상의 JSON 데이터 10건을 생성합니다.
IN 파라미터:
  dept_id: 10, 타입: <class 'int'>
  status: ACTIVE, 타입: <class 'str'>
OUT 파라미터:
  emp_names: []
  emp_names 길이: 0
  emp_names 타입: <class 'list'>
  emp_names 열 타입: [('EMPLOYEE_ID', 'NUMBER'), ('EMAIL', 'VARCHAR2'), ('BIRTH_DATE', 'DATE'), ('DEPARTMENT_NAME', 'VARCHAR2'), ('NOTES', 'CLOB')]
  emp_count: 0
  emp_count 타입: <class 'float'>
  status_message: No employees found for dept_id 10 and status ACTIVE
  status_message 타입: <class 'str'>
JSON 데이터: [
  {
    "EMPLOYEE_ID": 1234,
    "EMAIL": "sarah.taylor@example.com",
    "BIRTH_DATE": "1995-07-15",
    "DEPARTMENT_NAME": "Tech Solutions",
    "NOTES": "Highly skilled in project management."
  },
  ...
  {
    "EMPLOYEE_ID": 5678,
    "EMAIL": "robert.harris@example.com",
    "BIRTH_DATE": "1988-02-20",
    "DEPARTMENT_NAME": "Innovatech",
    "NOTES": "Received employee of the month award."
  }
]

시나리오 3: 최소 정보 (dept_id=10, status='ACTIVE', 0건)

 

시나리오 3: 최소 정보 (dept_id=10, status='ACTIVE', 0개 행)
emp_names가 비어있으므로 가상의 JSON 데이터 10건을 생성합니다.
IN 파라미터:
  dept_id: 10, 타입: <class 'int'>
  status: ACTIVE, 타입: <class 'str'>
OUT 파라미터:
  emp_names: []
  emp_names 길이: 0
  emp_names 타입: <class 'list'>
  emp_names 열 타입: [('FULL_NAME', 'VARCHAR2'), ('IS_ACTIVE', 'CHAR')]
  emp_count: 0
  emp_count 타입: <class 'float'>
  status_message: No employees found for dept_id 10 and status ACTIVE
  status_message 타입: <class 'str'>
JSON 데이터: [
  {
    "FULL_NAME": "Emma White",
    "IS_ACTIVE": "Y"
  },
  ...
  {
    "FULL_NAME": "James Jackson",
    "IS_ACTIVE": "N"
  }
]

---

코드 특징 및 개선점

1. 다양한 타입 지원:
   • VARCHAR2, NUMBER, DATE, CHAR, CLOB를 동적으로 처리.
   • 추가 타입(예: TIMESTAMP, BINARY_DOUBLE) 지원 가능.
2. 0개 행 처리:
   • get_column_metadata로 SELECT ... WHERE 1=0 실행, 메타데이터 추출.
   • 쿼리는 시나리오별로 scenario_queries에 정의.
3. 현실적 데이터:
   • faker로 이름, 이메일, 회사명, 텍스트 생성.
   • salary, birth_date, status_code 등 필드별 맞춤 데이터.
4. 유연성:
   • metadata_query만 변경하면 어떤 열 구조든 처리 가능.
   • generate_dummy_json_data는 필드 이름과 타입에 따라 동적으로 작동.

---

한계 및 대안

1. 쿼리 정의 필요:
   • metadata_query는 프로시저의 SELECT 문과 동일해야 함.
   • 대안: 프로시저 소스 코드를 파싱하거나, DBMS_SQL.DESCRIBE_COLUMNS 사용 (복잡).
2. 동적 SQL 문제:
   • 프로시저가 조건문으로 다른 SELECT 문을 실행하면 metadata_query가 부정확할 수 있음.
   • 대안: 프로시저에서 메타데이터를 별도 OUT 파라미터로 반환.
3. 성능:
   • 별도 쿼리 실행은 미미한 오버헤드 발생.
   • 대안: 메타데이터 캐싱.

---

주의사항

1. Faker 설치: pip install faker.
2. 쿼리 일치: scenario_queries의 쿼리는 프로시저의 SELECT 문과 열 구조가 동일해야 함.
3. 데이터베이스 환경:
   • employees 테이블과 열이 존재해야 함.
   • db_config에 실제 연결 정보 입력.
4. CLOB 처리:
   • CLOB은 200자로 제한. 대용량 데이터는 별도 처리 필요.

---

마무리

이 코드는 emp_names의 다양한 열 타입(VARCHAR2, NUMBER, DATE, CHAR, CLOB)을 동적으로 처리하며, 0개 행일 때 SELECT ... WHERE 1=0 쿼리로 메타데이터를 추출합니다. 세 가지 시나리오(기본, 상세, 최소 정보)를 통해 유연성을 입증했습니다. 추가 요구사항(예: 다른 타입, 동적 쿼리 파싱)이 있다면 알려주세요!