你能教我如何将 GCP 中的云函数连接到云 SQL 吗？

Question

看到了GCP快速入门(https://cloud.google.com/sql/docs/mysql/quickstart-connect-functions)，尝试将谷歌云函数连接到云SQL的My SQL。实际上该网站正在使用 HTTP 触发器，但我想使用 GCS 中的 “Finalize/Create”操作作为云函数的触发器。因此我使用了以下python代码和txt代码。

【MAIN.PY】

import sqlalchemy
from google.cloud import storage

# Set the following variables depending on your specific
# connection name and root password from the earlier steps:


def sql_select(data, context):
    connection_name = "INSTANCE_CONNECTION_NAME"
    db_password = "DATABASE_USER_PASSWORD"
    db_name = "DATABASE_NAME"
    db_user = "root" 
    driver_name = 'mysql+pymysql'
    query_string = '"unix_socket": "/cloudsql/{}".format(connection_name)'
    print("I am here 1")

    db = sqlalchemy.create_engine(
        sqlalchemy.engine.url.URL(
            drivername=driver_name,
            username=db_user,
            password=db_password,
            database=db_name,
            query={query_string},
        ),
        pool_size=5,
        max_overflow=2,
        pool_timeout=30,
        pool_recycle=1800)
    

    # stmt = sqlalchemy.text('INSERT INTO entries (guestName, content) values ("third guest", "Also this one");')
    # try:
    #     with db.connect() as conn:
    #         conn.execute(stmt)
    # except Exception as e:
    #     return 'Error: {}'.format(str(e))
    # return 'ok'

【需求.TXT】

SQLAlchemy==1.3.18

    PyMySQL==0.9.3

Click==7.0
Flask==1.0.2
itsdangerous==1.1.0
Jinja2==2.10
MarkupSafe==1.1.0
Pillow==5.4.1
qrcode==6.1
six==1.12.0
Werkzeug==0.14.1
google-cloud-storage==1.23.0

我尝试通过按下“测试功能”选项卡来测试代码。但是我有以下错误，无法将我制作的云功能连接到云 SQL(My SQL)。

【错误信息】

Error: function terminated. Recommended action: inspect logs for termination reason. Details:
'set' object has no attribute 'get' 

我尝试更改为最新版本的 sqlalchemy。但我无法解决错误。 我不知道要解决这个问题。 是否有任何解决方案可以将云功能连接到云 SQL？ 谢谢

Answer 1

您引用的链接应该为您提供几乎所需的一切。

您应该（！？）能够触发 GCS 存储桶事件的函数，但您需要确保正确部署该函数：

见：https://cloud.google.com/functions/docs/calling/storage#object_finalize

您需要import sqlalchemy。您已将包包含在 requirements.txt 中，但错过了第二步。

在创建 Cloud SQL 实例、数据库等之后，您需要更新 connection_name、db_password 等的值。

我认为您的query_string 陈述不正确，最好是：

query_string = dict({"unix_socket": "/cloudsql/{}".format(connection_name)})

这可能是错误的根源。