반응형
그냥 pymysql로 접속했을 때 로컬에서는 잘 작동했다. 하지만 앱엔진에 올리고 나면 연결 오류가 났다.
1. 장고Django 내 settings.py에서 설정할 경우
정상 접속되었다.
MYSQL_HOST: "/cloudsql/<PROJECT-NAME>:<INSTANCE-REGION>:<INSTANCE-NAME>"
로 해서 app.yaml에 추가했다.
if os.getenv('GAE_APPLICATION', None):
# Running on production App Engine, so connect to Google Cloud SQL using
# the unix socket at /cloudsql/<your-cloudsql-connection string>
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.mysql',
'HOST': os.getenv('MYSQL_HOST'),
'USER': os.getenv('MYSQL_USER'),
'PASSWORD': os.getenv('MYSQL_PASSWORD'),
'NAME': os.getenv('MYSQL_DB'),
}
}
else:
# Running locally so connect to either a local MySQL instance or connect to
# Cloud SQL via the proxy. To start the proxy via command line:
#
# $ cloud_sql_proxy -instances=[INSTANCE_CONNECTION_NAME]=tcp:3306
#
# See https://cloud.google.com/sql/docs/mysql-connect-proxy
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.mysql',
'HOST': '127.0.0.1',
'PORT': '3306',
'NAME': os.getenv('MYSQL_DB'),
'USER': os.getenv('MYSQL_USER'),
'PASSWORD': os.getenv('MYSQL_PASSWORD'),
}
}
# [END db_setup]
2. 따로 pymysql 사용한 경우
연결 오류가 났다.
참고: github.com/PyMySQL/PyMySQL
pymysql github 페이지에 나온 예대로 그대로 했는데도 계속 오류가 났다.
로컬에서는 잘 되는데 앱 엔진에 올리면 꼭 오류가 났다.
3. sqlalchemy 사용한 경우
정상 접속되었다.
참고: App Engine에서 Cloud SQL로 연결
참고: Github GoogleCloudPlatform: python-docs-samples/cloud-sql/mysql/sqlalchemy/
위 코드를 살펴보니 sqlalchemy로 예제가 나와 있었다.
그래서 sqlalchemy로 코드를 바꿔서 실행하니 정상 연결되었다.
정상 접속 테스트를 위한 코드
임의로 테스트 테이블을 하나 만들고, 거기에 입력해보도록 했다.
from flask import Flask, render_template, request, Response
import sqlalchemy
import os
import logging
app = Flask(__name__)
logger = logging.getLogger()
def init_connection_engine():
db_config = {
# [START cloud_sql_mysql_sqlalchemy_limit]
# Pool size is the maximum number of permanent connections to keep.
"pool_size": 5,
# Temporarily exceeds the set pool_size if no connections are available.
"max_overflow": 2,
# The total number of concurrent connections for your application will be
# a total of pool_size and max_overflow.
# [END cloud_sql_mysql_sqlalchemy_limit]
# [START cloud_sql_mysql_sqlalchemy_backoff]
# SQLAlchemy automatically uses delays between failed connection attempts,
# but provides no arguments for configuration.
# [END cloud_sql_mysql_sqlalchemy_backoff]
# [START cloud_sql_mysql_sqlalchemy_timeout]
# 'pool_timeout' is the maximum number of seconds to wait when retrieving a
# new connection from the pool. After the specified amount of time, an
# exception will be thrown.
"pool_timeout": 30, # 30 seconds
# [END cloud_sql_mysql_sqlalchemy_timeout]
# [START cloud_sql_mysql_sqlalchemy_lifetime]
# 'pool_recycle' is the maximum number of seconds a connection can persist.
# Connections that live longer than the specified amount of time will be
# reestablished
"pool_recycle": 1800, # 30 minutes
# [END cloud_sql_mysql_sqlalchemy_lifetime]
}
if os.environ.get("DB_HOST"):
print("DB_HOST")
return init_tcp_connection_engine(db_config)
else:
return init_unix_connection_engine(db_config)
def init_tcp_connection_engine(db_config):
# [START cloud_sql_mysql_sqlalchemy_create_tcp]
# Remember - storing secrets in plaintext is potentially unsafe. Consider using
# something like https://cloud.google.com/secret-manager/docs/overview to help keep
# secrets secret.
db_user = os.environ["DB_USER"]
db_pass = os.environ["DB_PASS"]
db_name = os.environ["DB_NAME"]
db_host = os.environ["DB_HOST"]
print(db_user)
print(db_pass)
print(db_name)
print(db_host)
# Extract host and port from db_host
host_args = db_host.split(":")
db_hostname, db_port = host_args[0], int(host_args[1])
pool = sqlalchemy.create_engine(
# Equivalent URL:
# mysql+pymysql://<db_user>:<db_pass>@<db_host>:<db_port>/<db_name>
sqlalchemy.engine.url.URL(
drivername="mysql+pymysql",
username=db_user, # e.g. "my-database-user"
password=db_pass, # e.g. "my-database-password"
host=db_hostname, # e.g. "127.0.0.1"
port=db_port, # e.g. 3306
database=db_name, # e.g. "my-database-name"
),
**db_config
)
# [END cloud_sql_mysql_sqlalchemy_create_tcp]
return pool
def init_unix_connection_engine(db_config):
# [START cloud_sql_mysql_sqlalchemy_create_socket]
# Remember - storing secrets in plaintext is potentially unsafe. Consider using
# something like https://cloud.google.com/secret-manager/docs/overview to help keep
# secrets secret.
db_user = os.environ["DB_USER"]
db_pass = os.environ["DB_PASS"]
db_name = os.environ["DB_NAME"]
db_socket_dir = os.environ.get("DB_SOCKET_DIR", "/cloudsql")
cloud_sql_connection_name = os.environ["CLOUD_SQL_CONNECTION_NAME"]
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(
drivername="mysql+pymysql",
username=db_user, # e.g. "my-database-user"
password=db_pass, # e.g. "my-database-password"
database=db_name, # e.g. "my-database-name"
query={
"unix_socket": "{}/{}".format(
db_socket_dir, # e.g. "/cloudsql"
cloud_sql_connection_name) # i.e "<PROJECT-NAME>:<INSTANCE-REGION>:<INSTANCE-NAME>"
}
),
**db_config
)
# [END cloud_sql_mysql_sqlalchemy_create_socket]
return pool
@app.route('/')
def hello_world():
return 'Hello World!'
db = None
@app.route('/test')
def test():
global db
db = db or init_connection_engine()
stmt = sqlalchemy.text(
"INSERT INTO test (id, username) VALUES (1, 'test')"
)
try:
with db.connect() as conn:
conn.execute(stmt)
except Exception as e:
logger.exception(e)
return 'error'
return 'test'
if __name__ == '__main__':
app.run()
반응형