如何通过Python将PostgreSQL查询结果转换为JSON对象

1. 先看优化后的完整代码 #

import psycopg2
import psycopg2.extras
import json
from botocore.exceptions import ClientError
import boto3

def get_db_credentials():
    # 替换成你的Secrets Manager凭证ARN和区域
    secret_name = "your-db-secret-arn"
    region_name = "us-east-1"
    
    session = boto3.session.Session()
    client = session.client(service_name='secretsmanager', region_name=region_name)
    
    try:
        get_secret_value_response = client.get_secret_value(SecretId=secret_name)
    except ClientError as e:
        raise e
    
    secret = json.loads(get_secret_value_response['SecretString'])
    return secret

def lambda_handler(event, context):
    db_credentials = get_db_credentials()
    connection = None
    cursor = None
    try:
        # 建立数据库连接
        connection = psycopg2.connect(
            host=db_credentials['host'],
            database=db_credentials['dbname'],
            user=db_credentials['username'],
            port=db_credentials['port'],
            password=db_credentials['password']
        )
        
        # 使用DictCursor直接返回带列名的字典，方便转JSON
        cursor = connection.cursor(cursor_factory=psycopg2.extras.DictCursor)
        
        # 后续可从event取参数，先保留硬编码示例
        table_to_select = 'fps.tbl_contractor'
        query_field = 'contractor_id'
        record_to_select = 101
        
        # 安全处理表名/字段名（标识符），避免SQL注入
        quoted_table = psycopg2.extensions.quote_ident(table_to_select, connection)
        quoted_field = psycopg2.extensions.quote_ident(query_field, connection)
        
        # 参数化查询，绝对不能用f-string拼值
        sql_select_query = f'SELECT * FROM {quoted_table} WHERE {quoted_field} = %s'
        cursor.execute(sql_select_query, (record_to_select,))
        
        # 获取结果并转成JSON格式
        results = cursor.fetchall()
        if not results:
            return {"Error": "Record not found or does not exist!"}, 404
        
        json_results = [dict(row) for row in results]
        
        # 返回符合API规范的JSON响应
        return {
            "statusCode": 200,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps(json_results)
        }
        
    except (Exception, psycopg2.Error) as error:
        print(f"Error in operation: {error}")
        return {"Error": "Internal server error"}, 500
    
    finally:
        # 确保资源关闭
        if cursor is not None:
            cursor.close()
        if connection is not None:
            connection.close()
            print("PostgreSQL connection is closed")

2. 关键问题拆解与解决 #

（1）彻底消灭SQL注入风险 #

• 参数化查询：字段值（比如contractor_id=101）必须用%s占位符，把参数放在execute的第二个参数里，psycopg2会自动转义，绝对不能用f-string拼值。
• 标识符转义：表名、字段名这类数据库标识符不能参数化，必须用psycopg2.extensions.quote_ident处理，防止恶意输入（比如fps.tbl_contractor; DROP TABLE...）。

（2）把查询结果转成JSON返回 #

• 用psycopg2.extras.DictCursor：查询结果直接是带列名的字典，不用手动映射列和值，转JSON一步到位。
• 响应格式：返回时设置Content-Type: application/json，用json.dumps把字典列表序列化后放在body里，前端能直接解析。

（3）代码安全与规范优化 #

• 移除无用的commit：SELECT操作不需要提交事务，只有写操作（INSERT/UPDATE/DELETE）才需要commit。
• 用Secrets Manager存凭证：绝对不能把数据库密码硬编码在代码里，AWS Secrets Manager可以安全存储、自动轮转凭证。
• 错误分级处理：区分“记录不存在”（404）和“服务器内部错误”（500），不要所有错误都返回404。

（4）适配后续接收JSON负载的需求 #

当需要从前端接收表名、查询字段和值时，替换硬编码部分即可：

# 解析前端传入的JSON负载
payload = json.loads(event['body'])
table_to_select = payload['table_name']
query_field = payload['query_field']
record_to_select = payload['value']

注意：一定要加白名单校验，只允许访问你指定的表和字段，防止用户越权访问数据库。

（5）关于PostgreSQL的to_json函数 #

你之前查的to_json也能用，直接在SQL里返回JSON：

SELECT to_json(t) FROM (SELECT * FROM fps.tbl_contractor WHERE contractor_id = %s) t

但用DictCursor在Python层处理更灵活，适合需要对结果做额外过滤、转换的场景。

内容的提问来源于stack exchange，提问作者Paul Jones