У меня возникла та же проблема, и после долгих поисков в Google это наконец-то решило ее для меня.
Пример показывает следующее предложение try:
if not mysql_conn:
try:
mysql_conn = pymysql.connect(**mysql_config)
except OperationalError:
# If production settings fail, use local development ones
mysql_config['unix_socket'] = f'/cloudsql/{CONNECTION_NAME}'
mysql_conn = pymysql.connect(**mysql_config)
Я изменилэто:
if not mysql_conn:
mysql_config['unix_socket'] = f'/cloudsql/{CONNECTION_NAME}'
mysql_conn = pymysql.connect(**mysql_config)
И использовал (data, context) в качестве аргументов функции, а не (request)
Это то, что я в конечном итоге закончил как тестовая функция:
from os import getenv
import pymysql
from pymysql.err import OperationalError
CONNECTION_NAME = getenv(
'INSTANCE_CONNECTION_NAME',
'connection_name')
DB_USER = getenv('MYSQL_USER', 'username')
DB_PASSWORD = getenv('MYSQL_PASSWORD', 'password')
DB_NAME = getenv('MYSQL_DATABASE', 'database')
mysql_config = {
'user': DB_USER,
'password': DB_PASSWORD,
'db': DB_NAME,
'charset': 'utf8mb4',
'cursorclass': pymysql.cursors.DictCursor,
'autocommit': True
}
if getenv('NODE_ENV', '') == 'production':
mysql_config['unix_socket'] = f'/cloudsql/{CONNECTION_NAME}'
# Create SQL connection globally to enable reuse
# PyMySQL does not include support for connection pooling
mysql_conn = None
def __get_cursor():
"""
Helper function to get a cursor
PyMySQL does NOT automatically reconnect,
so we must reconnect explicitly using ping()
"""
try:
return mysql_conn.cursor()
except OperationalError:
mysql_conn.ping(reconnect=True)
return mysql_conn.cursor()
def mysql_demo(data, context):
global mysql_conn
# Initialize connections lazily, in case SQL access isn't needed for this
# GCF instance. Doing so minimizes the number of active SQL connections,
# which helps keep your GCF instances under SQL connection limits.
if not mysql_conn:
mysql_conn = pymysql.connect(**mysql_config)
# Remember to close SQL resources declared while running this function.
# Keep any declared in global scope (e.g. mysql_conn) for later reuse.
with __get_cursor() as cursor:
cursor.execute('SELECT NOW() as now')
results = cursor.fetchone()
return str(results['now'])