Cloud Functionsから直接接続できます。（CloudSQL Direct Connect）

公式には、Node.js版しか記載されていませんが、GCP Next SF 2018 でβになったpython 3.7でも可能なので、ポイントを記載します。

公式のCloudSQL Direct Connectは以下を参照してください。

https://cloud.google.com/functions/docs/sql

functionsデプロイ時に必要なモジュールを記載。（mysql-connector-python）

requirements.txt

 google-cloud-logging==1.6.0
mysql-connector-python==8.0.11

次にコード

CloudSQL接続情報とかをまず設定します。unix socketに対して接続する設定になります。

import mysql.connector

connectionName = 'gcp-project-id:asia-east1:web2-db'
config = {
'user': 'dbuser',
'password': 'xxxxxx',
'unix_socket': '/cloudsql/' + connectionName,
'database': 'database'
}

Cloud Functions python 3.7 runtimeは、Flask microframeworkというフレームワークでコーディングします。

 def hello_sql(request):
"""HTTP Cloud Function.
Args:
request (flask.Request): The request object.
Returns:
The response text, or any set of values that can be tuned into a
Response object using `make_response
<http://flask.pocoo.org/docs/0.12/api/#flask.Flask.make_response>`.
"""
request_json = request.get_json()
if request_json and 'exec_cmd' in request_json:
exec_cmd = request_json['exec_cmd']

ここで、ポイント

Connection Poolに接続するのが、max connection を1にすること

Functionsはイベントトリブンなので、コネクションリソースを枯渇することを防ぐ上で。

公式にも以下記載されています。

When using a connection pool, it is important to set the maximum connections to 1. This may seem counter-intuitive, however, creating more than one concurrent connection per function instance may cause rapid exhaustion of connection resources (see Maximum Concurrent Connections below for more detail). Cloud Functions limits concurrent executions to 1 per instance. This means you will never have a situation where two requests are being processed by a single function instance at the same time, so in most situations only a single database connection is needed.

try:
cnx = mysql.connector.connect(pool_name="mypool",
pool_size = 1,
**config)
exec_cmd = cnx.is_connected()
if cnx.is_connected():
cur = cnx.cursor()
cur.execute('select * from wp_options limit 1 ')
result = cur.fetchall()
except:
exec_cmd = cnx.is_connected()
finally:
cur.close()
cnx.close()

return f'Hello, {result}!' 

αテスト時は、event handlingがでなくても、デバッグしづらかったのですが、最近、python コードのエラーも場合でも、functionsがcrashした場合でも、Trackbackが出るようになっています。