如何使用 python 在云函数中返回 postgreSQL 查询结果？ [复制]答案

【问题标题】：How to return postgreSQL query result in cloud function using python? [duplicate]如何使用 python 在云函数中返回 postgreSQL 查询结果？ [复制]
【发布时间】：2019-11-13 15:08:09
【问题描述】：

我正在使用 python 3.6 编写一个带有 get 方法的烧瓶 api，它将创建与 GCP postgreSQL 的连接并执行一个简单的选择查询。在这里，我希望我的函数应该返回一个 postgreSQL 查询（行）的结果。下面是我的代码：

from sqlalchemy.engine import url
import pg8000
import logging
import os
import json
from flask import Flask, Response, request, render_template
import sqlalchemy

app = Flask(__name__)

logger = logging.getLogger()

# Environment specific parameter values to establish DB connection with postgreSQL
db_user = os.environ.get("DB_USER")
db_pass = os.environ.get("DB_PASS")
db_name = os.environ.get("DB_NAME")
cloud_sql_connection_name = os.environ.get("CLOUD_SQL_CONNECTION_NAME")

# Creating a connection with postgresql DB model.
db = sqlalchemy.create_engine(
        url.URL(
            drivername='postgres+pg8000',
            username=db_user,
            password=db_pass,
            database=db_name,
            query={
                'unix_sock': '/cloudsql/{}/.s.PGSQL.5432'.format(
                    cloud_sql_connection_name)
            }
        ),
        pool_size=5,
        max_overflow=2,
        pool_timeout=30,
        pool_recycle=1800,
    )


@app.route('/', methods=['GET'])
def main(request):

    with db.connect() as conn:
        # Execute the query and fetch all usernames
        users = conn.execute("select username from user_table").fetchall()

    return users


if __name__ == '__main__':
    app.run(host='127.0.0.1', port=8080, debug=True)

【问题讨论】：

标签： python flask

【解决方案1】：

我尝试了您的代码并收到此错误：

TypeError：视图函数未返回有效响应。返回类型必须是字符串、字典、元组、响应实例或 WSGI 可调用，但它是一个列表

所以尝试返回str(users) 或先将数据转换为dict。

【讨论】：

【解决方案2】：

fetchall() 从文档https://docs.sqlalchemy.org/en/13/core/connections.html#sqlalchemy.engine.ResultProxy.fetchall 返回光标。您不能将光标对象作为响应返回。您需要将其转换为允许的响应类型。下面的代码可能会帮助您理解相同的内容。


@app.route('/', methods=['GET'])
def main(request):
    users = []
    with db.connect() as conn:
        # Execute the query and fetch all usernames
        db_users = conn.execute("select username from user_table").fetchall()
        users = list(db_users)
    data = {'users': users}
    resp = Response(json.dumps(data), status=200, mimetype="application/json")
    return resp

【讨论】：