Google Flex App Engine无法写入Cloud SQL数据库问题及解决

你遇到的这个情况很典型——本地测试一切正常，但部署到App Engine Flex后，明明Cloud SQL连接正常、能收到Meraki的POST请求，数据库和表结构也都存在，就是INSERT语句不生效。结合你的代码和配置，我来梳理下问题的核心和最终解决方案：

场景回顾 #

你基于Flask+SQLAlchemy开发了接收Meraki Scanning API数据的服务，本地MySQL能正常写入数据，但部署到App Engine Flex后，数据始终无法写入Cloud SQL。已经确认：

• App Engine与Cloud SQL的连接已建立
• 数据库及表结构均存在
• 能正常接收Meraki的POST请求

关键代码片段 #

数据写入逻辑

from api import app
from sqlalchemy import create_engine
from datetime import datetime, timedelta

engine = create_engine(app.config.get('database_uri'))

class Events(object):
    @staticmethod
    def add(event):
        connection = engine.connect()
        for observation in event["observations"]:
            try:
                connection.execute("""INSERT INTO events ( 
                    apMac, apTags, apFloors, clientMac, ipv4, ipv6, seenTime, seenEpoch, ssid, rssi, manufacturer, os, lat, lng, unc, x, y 
                ) VALUES ( 
                    %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s 
                )""", ( 
                    event.get("apMac", None),
                    ",".join(event.get("apTags", [])),
                    ",".join(event.get("apFloors", [])),
                    observation.get("clientMac", None),
                    observation.get("ipv4").strip("/") if observation.get("ipv4") is not None else None,
                    observation.get("ipv6", None),
                    observation.get("seenTime", None),
                    observation.get("seenEpoch", None),
                    observation.get("ssid", None),
                    observation.get("rssi", None),
                    observation.get("manufacturer", None),
                    observation.get("os", None),
                    observation.get("location", {}).get("lat", None),
                    observation.get("location", {}).get("lng", None),
                    observation.get("location", {}).get("unc", None),
                    observation.get("location", {}).get("x", None) if observation.get("location", {}).get("x", None) != [] else None,
                    observation.get("location", {}).get("y", None) if observation.get("location", {}).get("y", None) != [] else None,
                ))
            except:
                pass
        connection.close()
        return "Added"

请求处理逻辑

from flask import request
from api import app
from api.models import Events

@app.route('/', methods=['GET'])
def events_get():
    return str(app.config.get("validator"))

@app.route('/', methods=['POST'])
def events_post():
    data = request.json
    if data["secret"] != app.config.get("secret"):
        return Events.add(data["data"])
    return "."

数据库URI配置

mysql+mysqldb://<user>:<pw>@/<database>?unix_socket=/cloudsql/<instance connection name>

最终解决方案 #

你提到的将Cloud SQL降级至MySQL 5.6，并开启MySQL的ALLOW_INVALID_DATES参数确实是解决问题的关键，背后的原因是：

• 高版本MySQL（5.7及以上）对DateTime类型的格式校验更为严格，而Meraki Scanning API返回的seenTime格式可能不符合高版本MySQL的默认校验规则
• 开启ALLOW_INVALID_DATES后，MySQL会跳过对日期格式的严格校验（仅检查月份在1-12、日期在1-31范围内），从而兼容Meraki返回的时间数据格式

额外排查建议 #

你的代码中使用了except:无差别捕获所有异常并直接pass，这会导致你完全看不到插入失败的具体错误信息，非常不利于问题排查。建议修改为捕获具体异常并记录日志：

except Exception as e:
    app.logger.error(f"插入观测数据失败: {str(e)}")
    # 可根据业务需求选择是否继续处理其他观测数据

这样下次遇到类似问题时，你可以直接从App Engine的日志中看到具体错误（比如时间格式不兼容、字段长度超限等），排查效率会大幅提升。

内容的提问来源于stack exchange，提问作者jhomr