在Google Cloud Run的Flask应用中连接Cloud Postgres的最简方法

psycopg2依赖PostgreSQL的系统底层库（如libpq），Cloud Run的基础Python镜像默认缺少这些依赖，所以部署后会报错。pg8000是纯Python实现的PostgreSQL驱动，无需系统依赖，更适合容器化部署场景。以下是最简的连接与查询实现：

1. 依赖准备 #

在项目的requirements.txt中添加：

pg8000>=1.29.0
Flask>=2.0.0

2. 最简连接查询函数 #

import pg8000.native

def query_postgres(query):
    # 替换为你的数据库实际参数
    db_config = {
        "user": "postgres",
        "password": "password",
        "host": "Public IP Address",
        "port": 5432,
        "database": "postgres"
    }
    
    # 自动管理连接生命周期
    with pg8000.native.Connection(**db_config) as conn:
        # 执行查询并转换结果为列表
        result = list(conn.run(query))
    return result

3. Flask中使用示例 #

from flask import Flask

app = Flask(__name__)

@app.route("/")
def fetch_data():
    # 替换为你的实际查询语句
    data = query_postgres("SELECT * FROM your_table LIMIT 10")
    return {"result": data}

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

关键提示 #

• 避免硬编码敏感信息：在Cloud Run中通过环境变量存储密码、用户名等，示例：

  import os
  db_config = {
      "user": os.getenv("DB_USER"),
      "password": os.getenv("DB_PASSWORD"),
      "host": os.getenv("DB_HOST"),
      "port": int(os.getenv("DB_PORT", 5432)),
      "database": os.getenv("DB_NAME")
  }
  

• 参数化查询防注入：如果需要动态传参，用占位符%s：

  # 查询id=5的记录
  result = list(conn.run("SELECT * FROM your_table WHERE id = %s", (5,)))
  

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