从 Google Functions (PY) 连接到 MySQL SQL Server

Question

我正在尝试从 Python 3.11 中的 Google Functions 连接到我已设置的 MySQL 实例。

@functions_framework.http def mysql_engine(请求): 选项 = request.get_json() MYSQL_USER = 选项['用户'] MYSQL_PASSWORD = 选项['pass'] MYSQL_HOST = 选项['主机'] MYSQL_DB_NAME = 选项['db_name']

engine_gam = sqlalchemy.create_engine(
    'mysql+mysqlconnector://' +
    MYSQL_USER +
    ':' +
    MYSQL_PASSWORD +
    '@' +
    MYSQL_HOST +
    '/' +
    MYSQL_DB_NAME
)
engine_gam.connect()

但是，我收到以下错误。

sqlalchemy.exc.InterfaceError：（mysql.connector.errors.InterfaceError） 2003：无法连接到“x.x.x.x:xxxx”上的 MySQL 服务器（110 连接 超时）（此错误的背景位于： https://sqlalche.me/e/20/rvf5）

我知道“简单”的解决方案是在 SQL Server 中添加 0.0.0.0/0 作为白名单，并且它可以使用此方法，但是我想知道是否可以使用服务帐户作为白名单方法.

Google 功能正在使用的服务帐户：

IAM 中的相同服务帐户权限：

这就是我目前的设置方式，我是否遗漏了什么？

Answer 1

对于尝试使用 MySQL 用户和密码进行连接的最基本用例，请尝试使用内置 Cloud SQL 代理进行连接。请按照以下说明进行操作，这些说明描述了如何在 Python 中配置 unix 套接字路径。

通过这种方式，您无需在应用程序中安装任何其他库即可进行连接。

import os

import sqlalchemy


def connect_unix_socket() -> sqlalchemy.engine.base.Engine:
    """Initializes a Unix socket connection pool for a Cloud SQL instance of MySQL."""
    # Note: Saving credentials in environment variables is convenient, but not
    # secure - consider a more secure solution such as
    # Cloud Secret Manager (https://cloud.google.com/secret-manager) to help
    # keep secrets safe.
    db_user = os.environ["DB_USER"]  # e.g. 'my-database-user'
    db_pass = os.environ["DB_PASS"]  # e.g. 'my-database-password'
    db_name = os.environ["DB_NAME"]  # e.g. 'my-database'
    unix_socket_path = os.environ[
        "INSTANCE_UNIX_SOCKET"
    ]  # e.g. '/cloudsql/project:region:instance'

    pool = sqlalchemy.create_engine(
        # Equivalent URL:
        # mysql+pymysql://<db_user>:<db_pass>@/<db_name>?unix_socket=<socket_path>/<cloud_sql_instance_name>
        sqlalchemy.engine.url.URL.create(
            drivername="mysql+pymysql",
            username=db_user,
            password=db_pass,
            database=db_name,
            query={"unix_socket": unix_socket_path},
        ),
        # ...
    )
    return pool

但是，如果您计划使用 IAM 身份验证或通过私有 IP 地址连接到 MySQL 实例，则可能需要使用 Cloud SQL Connector for Python 库。

同一文档还展示了如何使用 Cloud Functions 中的

Cloud SQL Connectors 进行连接