Django中如何在不同视图内对远程Postgres发起查询

你现在的写法有两个明显问题：一是连接逻辑重复写在单个视图里，没法复用；二是没有处理多步流程里的凭证传递，还存在资源泄漏、SQL注入的隐患，按下面的步骤改就行。

1. 抽离公共连接方法 #

别在每个视图里重复写psycopg2连接代码，单独抽工具函数统一处理连接逻辑。你原来的代码只关了游标没关连接，跑久了会占满数据库连接池直接连不上，建议直接用上下文管理器自动释放资源：

import psycopg2
from psycopg2 import Error

def get_pg_connection(conn_params):
    """
    传入连接参数字典，返回pg连接对象
    conn_params 需包含 host/port/database/user/password 五个字段
    """
    try:
        return psycopg2.connect(**conn_params)
    except (Exception, Error) as e:
        raise ValueError(f"连接失败：{str(e)}")

重点提醒：用户输入的数据库凭证属于高敏感信息，禁止明文存在cookie或者前端本地存储，也不要做持久化保存，用完立刻销毁。

2. 第一步：接收凭证、拉取表列表 #

第一个视图对应你收数据库连接信息的模板，用户提交凭证后先测连通性，连通成功就拉取用户表列表，同时把加密后的连接参数存在服务端session里，跳转到表选择页面。
以Flask类视图为例，Django逻辑完全一致，替换session和请求对象的写法就行：

from flask import request, session, render_template, redirect

class DBConnectView:
    def get(self):
        return render_template("db_connect_form.html")
    
    def post(self):
        # 收集用户提交的连接参数
        conn_params = {
            "host": request.form.get("host"),
            "port": request.form.get("port"),
            "database": request.form.get("db_name"),
            "user": request.form.get("username"),
            "password": request.form.get("password")
        }

        try:
            # 测连通性，拉表列表
            with get_pg_connection(conn_params) as conn:
                with conn.cursor() as cur:
                    cur.execute("select relname from pg_class where relkind='r' and relname !~ '^(pg_|sql_)';")
                    table_list = [row[0] for row in cur.fetchall()]
            # 生产环境请先对conn_params做对称加密再存入session，避免明文泄露
            session["temp_pg_conn"] = conn_params
            return render_template("select_query.html", tables=table_list)
        except ValueError as e:
            return render_template("db_connect_form.html", error=str(e))

3. 第二步：执行自定义查询 #

第二个视图接收用户选的表、查询字段、筛选条件，这里绝对不能直接用字符串拼接SQL，会有严重SQL注入风险，用psycopg2自带的sql模块转义动态标识符，用参数化传值：

from psycopg2 import sql

class DataQueryView:
    def post(self):
        # 从session取连接参数，不存在就跳回第一步重填
        conn_params = session.pop("temp_pg_conn", None)
        if not conn_params:
            return redirect("/db-connect")
        
        target_table = request.form.get("selected_table")
        selected_cols = request.form.getlist("selected_columns")
        filter_val = request.form.get("filter_value") # 对应你表单里的查询条件值

        try:
            with get_pg_connection(conn_params) as conn:
                with conn.cursor() as cur:
                    # 安全构造动态SQL
                    if not selected_cols:
                        query = sql.SQL("SELECT * FROM {}").format(sql.Identifier(target_table))
                    else:
                        query = sql.SQL("SELECT {} FROM {}").format(
                            sql.SQL(', ').join(map(sql.Identifier, selected_cols)),
                            sql.Identifier(target_table)
                        )
                    # 拼接筛选条件，参数化传值
                    exec_params = ()
                    if filter_val:
                        query += sql.SQL(" WHERE id = %s") # 按你实际需要的筛选字段修改
                        exec_params = (filter_val,)
                    
                    cur.execute(query, exec_params)
                    # 拿列名，方便前端表格渲染
                    columns = [desc[0] for desc in cur.description]
                    query_result = cur.fetchall()
            
            return render_template("result.html", columns=columns, rows=query_result)
        except (Exception, Error) as e:
            # 查询失败把参数存回session，让用户可以重新选择查询条件
            session["temp_pg_conn"] = conn_params
            table_list = self._get_user_tables(conn_params)
            return render_template("select_query.html", tables=table_list, error=f"查询失败：{str(e)}")
    
    def _get_user_tables(self, conn_params):
        # 复用拉表列表的逻辑
        with get_pg_connection(conn_params) as conn:
            with conn.cursor() as cur:
                cur.execute("select relname from pg_class where relkind='r' and relname !~ '^(pg_|sql_)';")
                return [row[0] for row in cur.fetchall()]

几个必须注意的坑 #

• 所有动态传入的表名、字段名必须用sql.Identifier转义，禁止直接用f-string或者字符串拼接SQL，否则用户传恶意参数可以直接删库拖库
• 建议给用户连接的数据库账号只分配只读权限，从根源上避免恶意修改、删除数据的风险
• 存到session的连接参数一定要加密，给session cookie设置Secure、HttpOnly属性，降低凭证泄露风险
• 用with上下文管理器处理连接和游标，退出代码块会自动关闭资源，不会出现连接泄漏

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