Oracle数据库将表数据转储为JSON格式文件的详细步骤

def connect_to_oracle(host: str, port: int, service_name: str, username: str, password: str) -> cx_Oracle.Connection:
    """
    Establish connection to Oracle database
    
    Args:
        host: Database host
        port: Database port
        service_name: Service name
        username: Database username
        password: Database password
        
    Returns:
        cx_Oracle.Connection: Database connection object
    """
    dsn = cx_Oracle.makedsn(host, port, service_name=service_name)
    connection = cx_Oracle.connect(username, password, dsn)
    return connection

def fetch_data_from_table(connection: cx_Oracle.Connection, query: str) -> List[Dict[str, Any]]:
    """
    Fetch data from Oracle database using provided query
    
    Args:
        connection: Oracle database connection
        query: SQL query to execute
        
    Returns:
        List of dictionaries containing the query results
    """
    cursor = connection.cursor()
    cursor.execute(query)
    
    # Get column names
    columns = [desc[0].lower() for desc in cursor.description]
    
    # Fetch all rows
    rows = cursor.fetchall()
    
    # Convert to list of dictionaries
    result = []
    for row in rows:
        result.append(dict(zip(columns, row)))
    
    cursor.close()
    return result

def save_to_json(data: List[Dict[str, Any]], filename: str) -> None:
    """
    Save data to JSON file
    
    Args:
        data: Data to save
        filename: Output filename
    """
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2, default=str)
    print(f"Data saved to {filename}")

import cx_Oracle
import json
import os
from typing import List, Dict, Any

def connect_to_oracle(host: str, port: int, service_name: str, username: str, password: str) -> cx_Oracle.Connection:
    """
    Establish connection to Oracle database
    
    Args:
        host: Database host
        port: Database port
        service_name: Service name
        username: Database username
        password: Database password
        
    Returns:
        cx_Oracle.Connection: Database connection object
    """
    dsn = cx_Oracle.makedsn(host, port, service_name=service_name)
    connection = cx_Oracle.connect(username, password, dsn)
    return connection

def fetch_data_from_table(connection: cx_Oracle.Connection, query: str) -> List[Dict[str, Any]]:
    """
    Fetch data from Oracle database using provided query
    
    Args:
        connection: Oracle database connection
        query: SQL query to execute
        
    Returns:
        List of dictionaries containing the query results
    """
    cursor = connection.cursor()
    cursor.execute(query)
    
    # Get column names
    columns = [desc[0].lower() for desc in cursor.description]
    
    # Fetch all rows
    rows = cursor.fetchall()
    
    # Convert to list of dictionaries
    result = []
    for row in rows:
        result.append(dict(zip(columns, row)))
    
    cursor.close()
    return result

def save_to_json(data: List[Dict[str, Any]], filename: str) -> None:
    """
    Save data to JSON file
    
    Args:
        data: Data to save
        filename: Output filename
    """
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2, default=str)
    print(f"Data saved to {filename}")

def export_table_to_json(
    host: str, 
    port: int, 
    service_name: str, 
    username: str, 
    password: str,
    table_name: str,
    output_file: str,
    additional_conditions: str = ""
) -> None:
    """
    Export Oracle table data to JSON file
    
    Args:
        host: Database host
        port: Database port
        service_name: Service name
        username: Database username
        password: Database password
        table_name: Table name to export
        output_file: Output JSON filename
        additional_conditions: Additional WHERE conditions (optional)
    """
    try:
        # Connect to database
        connection = connect_to_oracle(host, port, service_name, username, password)
        print("Connected to Oracle database successfully")
        
        # Build query
        query = f"SELECT * FROM {table_name}"
        if additional_conditions:
            query += f" WHERE {additional_conditions}"
            
        print(f"Executing query: {query}")
        
        # Fetch data
        data = fetch_data_from_table(connection, query)
        print(f"Fetched {len(data)} records from database")
        
        # Save to JSON
        save_to_json(data, output_file)
        
        # Close connection
        connection.close()
        print("Database connection closed")
        
    except cx_Oracle.Error as e:
        print(f"Oracle error occurred: {e}")
    except Exception as e:
        print(f"An error occurred: {e}")

# Example usage
if __name__ == "__main__":
    # Database configuration - replace with your actual values
    DB_CONFIG = {
        'host': 'localhost',
        'port': 1521,
        'service_name': 'ORCL',
        'username': 'your_username',
        'password': 'your_password'
    }
    
    # Export parameters
    TABLE_NAME = 'your_table_name'
    OUTPUT_FILE = 'output_data.json'
    
    # Export data
    export_table_to_json(
        host=DB_CONFIG['host'],
        port=DB_CONFIG['port'],
        service_name=DB_CONFIG['service_name'],
        username=DB_CONFIG['username'],
        password=DB_CONFIG['password'],
        table_name=TABLE_NAME,
        output_file=OUTPUT_FILE
    )