Oracle数据库表数据转储为JSON格式文件
1. 参数
所需参数包含Oracle服务器连接参数、表名、输出文件路径
| 参数 | 含义 | 举例 |
|---|---|---|
| host | 服务器ip地址 | 如localhost、192.30.30.30 |
| port | 服务器端口号 | 默认1521 |
| service_name | 服务器名称 | 如ORACLE |
| username | 账号 | 如oracle_un |
| password | 密码 | 如oracle_ps |
| TABLE_NAME | 表名，即你需要将哪个表转储为json文件 | 如tb_oracle_data |
| OUTPUT_FILE | 输出文件名，json文件的保存位置和名称 | 如./output_data.json |
2. 数据库连接
首先需要连接数据库，以下为连接数据库的代码
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
3. 获取数据
获取表中的数据，通过连接参数查询表中字段和数据并整合
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
4. 保存JSON
将数据保存为json格式的文件输出
这里文件名可以自定义文件路径和名称
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}")
5. 完整代码
以下为完整代码
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
)
到此这篇关于Oracle数据库将表数据转储为JSON格式文件的详细步骤的文章就介绍到这了,更多相关Oracle表数据转储为JSON文件内容请搜索IT俱乐部以前的文章或继续浏览下面的相关文章希望大家以后多多支持IT俱乐部！
