平时做技术实践时，很多问题不是概念不会，而是细节没串起来。拿“Oracle数据库将表数据转储为JSON格式文件的详细步骤”来说，它看着像小点，放到项目里常会牵出环境、配置、兼容性和维护成本。下面按实际采用顺序，把思路、关键写法和容易踩坑的地方讲清楚，便于大家直接对照操作。
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文件内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多兼容脚本之家！
您可能感兴趣的文章:
- 采用zabbix监控oracle表数据的方法
- oracle delete误删除表数据后如何恢复
- oracle 批量删除表数据的几种方法
- Oracle误删除表数据后的数据恢复详解
- Oracle 跨库 查询 复制表数据 分布式查询介绍