K8s中Flask+GraphQL API无法正确获取PostgreSQL数据求助

在Kubernetes集群中部署了基于Flask框架、由uWSGI运行的GraphQL API，用于从PostgreSQL数据库获取数据。使用psycopg2的线程连接池，编写了两个Resolver分别查询不同表的数据。当UI同时发起多个查询时，多数情况能返回正确数据，但偶尔会出现数据互换、部分字段为null的异常。

PostgreSQL连接池代码 #

import psycopg2
from psycopg2 import pool
import psycopg2.extras

try:
    conn = pool.ThreadedConnectionPool(4, 8, 
                                       user='admin',
                                       password='password',
                                       host='100.66.19.56',
                                       database='db',
                                       cursor_factory = psycopg2.extras.RealDictCursor)
    ps_conn = conn.getconn()
    ps_conn.autocommit = True

except (Exception, psycopg2.DatabaseError) as error:
    print("Error in PostgreSQL Connection", error)

获取应用数据的Resolver #

import database.conn as pc
from graphql import GraphQLError

class FetchAppDataResolver:

    def resolve_FetchAppData(self, parent, info):
        res = []
        try:
            cursor1 = pc.ps_conn.cursor()
            cursor1.execute("SELECT * FROM db_schema.app_table;")
            app_details = cursor1.fetchall()
            cursor1.close()
            res.append({"status_msg": "Successful", "app_details": app_details})
            
        except(Exception, GraphQLError) as e:
            print("Error: ", str(e))
            pc.ps_conn.rollback()
            res.append({"status_msg": "Error", "app_details": []})
        return res

获取分类数据的Resolver #

import database.conn as pc
from graphql import GraphQLError

class FetchCatDataResolver:

    def resolve_FetchCatData(self, parent, info):
        res = []
        try:
            cursor2 = pc.ps_conn.cursor()
            cursor2.execute("SELECT * FROM db_schema.cat_table;")
            cat_details = cursor2.fetchall()
            cursor2.close()
            res.append({"status_msg": "Successful", "cat_details": cat_details})
            
        except(Exception, GraphQLError) as e:
            print("Error: ", str(e))
            pc.ps_conn.rollback()
            res.append({"status_msg": "Error", "cat_details": []})
        return res

异常响应示例 #

获取应用数据的异常响应 #

{
  "data": {
    "FetchAppDetails": [
      {
        "statusMsg": "Successful",
        "appDetails": [
          {
            "aId": null,
            "typeApp": null,
            "title": "Category1",
            "subTitle": null,
            "description": "Category1 desc"
          },
          {
            "aId": null,
            "typeApp": null,
            "title": "Category2",
            "subTitle": null,
            "description": "Category2 desc"
          }
        ]
      }
    ]
  }
}

获取分类数据的异常响应 #

{
  "data": {
    "FetchCatDetails": [
      {
        "statusMsg": "Successful",
        "catDetails": [
          {
            "cId": null,
            "typeCat": null,
            "title": "App1",
            "subCategory": null,
            "description": "App1 desc"
          },
          {
            "cId": null,
            "typeCat": null,
            "title": "App2",
            "subCategory": null,
            "description": "App2 desc"
          }
        ]
      }
    ]
  }
}

1. 全局共享数据库连接：连接池初始化后直接获取了全局的ps_conn对象，所有Resolver复用这个单一连接。在uWSGI多线程/多进程环境下，多个请求同时操作同一个连接会触发游标竞态：
   • 当A请求执行完查询但未获取结果时，B请求复用同一连接执行另一个查询，会覆盖前一个查询的结果集，导致数据混淆。
   • RealDictCursor按字段名映射结果，不同表字段不匹配时，不存在的字段会返回null，出现异常中的字段缺失情况。
2. 连接池使用错误：创建ThreadedConnectionPool后未按设计使用——每个线程/请求应独立从池内获取连接，用完归还，而非固定持有单个连接。

1. 修正连接池初始化逻辑 #

仅初始化连接池对象，不提前获取固定连接：

import psycopg2
from psycopg2 import pool
import psycopg2.extras

try:
    # 仅初始化连接池，不提前获取连接
    conn_pool = pool.ThreadedConnectionPool(4, 8, 
                                       user='admin',
                                       password='password',
                                       host='100.66.19.56',
                                       database='db',
                                       cursor_factory = psycopg2.extras.RealDictCursor)

except (Exception, psycopg2.DatabaseError) as error:
    print("Error in PostgreSQL Connection", error)

2. 为每个Resolver独立管理连接 #

修改Resolver代码，每次请求从连接池获取独立连接，执行完成后归还：

修正后的FetchAppDataResolver #

import database.conn as pc
from graphql import GraphQLError

class FetchAppDataResolver:

    def resolve_FetchAppData(self, parent, info):
        res = []
        ps_conn = None
        try:
            # 从连接池获取独立连接
            ps_conn = pc.conn_pool.getconn()
            ps_conn.autocommit = True
            cursor = ps_conn.cursor()
            cursor.execute("SELECT * FROM db_schema.app_table;")
            app_details = cursor.fetchall()
            cursor.close()
            res.append({"status_msg": "Successful", "app_details": app_details})
            
        except(Exception, GraphQLError) as e:
            print("Error: ", str(e))
            if ps_conn:
                ps_conn.rollback()
            res.append({"status_msg": "Error", "app_details": []})
        finally:
            # 用完归还连接到池
            if ps_conn:
                pc.conn_pool.putconn(ps_conn)
        return res

修正后的FetchCatDataResolver #

import database.conn as pc
from graphql import GraphQLError

class FetchCatDataResolver:

    def resolve_FetchCatData(self, parent, info):
        res = []
        ps_conn = None
        try:
            # 从连接池获取独立连接
            ps_conn = pc.conn_pool.getconn()
            ps_conn.autocommit = True
            cursor = ps_conn.cursor()
            cursor.execute("SELECT * FROM db_schema.cat_table;")
            cat_details = cursor.fetchall()
            cursor.close()
            res.append({"status_msg": "Successful", "cat_details": cat_details})
            
        except(Exception, GraphQLError) as e:
            print("Error: ", str(e))
            if ps_conn:
                ps_conn.rollback()
            res.append({"status_msg": "Error", "cat_details": []})
        finally:
            # 用完归还连接到池
            if ps_conn:
                pc.conn_pool.putconn(ps_conn)
        return res

3. 额外优化建议 #

• 调整uWSGI的线程/进程总数，避免超过连接池最大连接数（当前池最大8个连接，建议uWSGI workers+threads总数不超过8）。
• 为连接池添加connect_timeout=10等参数，避免闲置连接被数据库主动断开。
• 使用上下文管理器（with语句）自动管理连接和游标，简化代码并确保资源正确释放。

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