這是我的代碼的超級(jí)縮略版（我會(huì)在評(píng)論中添加指向所有內(nèi)容的鏈接）import ......def main(data, context): log_client = logging.Client() log_name = 'cloudfunctions.googleapis.com%2Fcloud-functions' res = Resource(type="cloud_function", labels={ "function_name": "refresh_classes", "region": os.environ.get("FUNC_REGION") }) logger = log_client.logger(log_name.format(os.environ.get("PROJECT_ID"))) db = sqlalchemy.create_engine( sqlalchemy.engine.url.URL( drivername="mysql+pymysql", username=os.environ.get("DB_USER"), password=os.environ.get("DB_PASS"), host=os.environ.get("DB_HOST"), port=3306, database=PRIMARY_TABLE_NAME ), pool_size=5, max_overflow=2, pool_timeout=30, pool_recycle=1800 ) start_time = perf_counter() check_if_table_exists(db) for i in range(START_IDX, END_IDX): print(i) logger.log_text(f"Checking class with id {i}", resource=res, severity="INFO") ... logger.log_text(f"Total seconds elapsed: {perf_counter() - start_time}", resource=res, severity="INFO")if __name__ == '__main__': main('data', 'context')當(dāng)我在本地運(yùn)行上述云函數(shù)時(shí)，我的GOOGLE_APPLICATION_CREDENTIALS配置與我的本地 Cloud MySQL 代理一起設(shè)置，詳細(xì)的云日志記錄通過(guò)并且該函數(shù)順利完成，
1 回答
呼喚遠(yuǎn)方
TA貢獻(xiàn)1856條經(jīng)驗(yàn) 獲得超11個(gè)贊
所以我終于想通了，如果不是通過(guò) Cloud Scheduler 觸發(fā)云功能，而是通過(guò)以下方式手動(dòng)運(yùn)行該功能Test Function
GCP 會(huì)給你引發(fā)的異常。就我而言，我的 Cloud MySQL 連接失敗
Error: function terminated. Recommended action: inspect logs for termination reason. Details:
(pymysql.err.OperationalError) (2003, "Can't connect to MySQL server on '**.***.**.***' (timed out)")
(Background on this error at: http://sqlalche.me/e/e3q8)
所以我只需要引入一個(gè)環(huán)境變量來(lái)在通過(guò)代理和unix 套接字連接之間切換，如下所示：
if os.environ.get("ENV") == "local":
db = sqlalchemy.create_engine(
sqlalchemy.engine.url.URL(
drivername="mysql+pymysql",
username=os.environ.get("DB_USER"),
password=os.environ.get("DB_PASS"),
host=os.environ.get("DB_HOST"),
port=3306,
database=PRIMARY_TABLE_NAME
),
pool_size=5,
max_overflow=2,
pool_timeout=30,
pool_recycle=1800
)
else:
db = sqlalchemy.create_engine(
sqlalchemy.engine.url.URL(
drivername="mysql+pymysql",
username=os.environ.get("DB_USER"),
password=os.environ.get("DB_PASS"),
database=PRIMARY_TABLE_NAME,
query={"unix_socket": "/cloudsql/{}".format(os.environ.get("CLOUD_SQL_CONNECTION_NAME"))}
),
pool_size=5,
max_overflow=2,
pool_timeout=30,
pool_recycle=1800
)
添加回答
舉報(bào)
0/150
提交
取消