在SQLAlchemy中,
mysql+gaerdbms:///
驱动程序只能使用
rdbms_apiproxy
DBAPI,而当从Google App Engine实例访问Google Cloud SQL时才能使用。我向SQLAlchemy提交了一个
工单,要求更新驱动程序,以便在不在Google App Engine上时使用基于OAuth的
rdbms_googleapi
,就像App Engine SDK提供的
Django驱动程序一样。rdbms_googleapi也是
google_sql.py
(远程SQL控制台)使用的DBAPI。
预计更新后的方言将成为0.7.10和0.8.0版本的一部分,但在它们可用之前,您可以执行以下操作:
1-将工单中的更新方言复制到文件中(例如gaerdbms_dialect.py)
from sqlalchemy.dialects.mysql.mysqldb import MySQLDialect_mysqldb
from sqlalchemy.pool import NullPool
import re
"""Support for Google Cloud SQL on Google App Engine
Connecting
-----------
Connect string format::
mysql+gaerdbms:///<dbname>?instance=<project:instance>
# Example:
create_engine('mysql+gaerdbms:///mydb?instance=myproject:instance1')
"""
class MySQLDialect_gaerdbms(MySQLDialect_mysqldb):
@classmethod
def dbapi(cls):
from google.appengine.api import apiproxy_stub_map
if apiproxy_stub_map.apiproxy.GetStub('rdbms'):
from google.storage.speckle.python.api import rdbms_apiproxy
return rdbms_apiproxy
else:
from google.storage.speckle.python.api import rdbms_googleapi
return rdbms_googleapi
@classmethod
def get_pool_class(cls, url):
return NullPool
def create_connect_args(self, url):
opts = url.translate_connect_args()
opts['dsn'] = ''
opts['instance'] = url.query['instance']
return [], opts
def _extract_error_code(self, exception):
match = re.compile(r"^(\d+):").match(str(exception))
code = match.group(1)
if code:
return int(code)
dialect = MySQLDialect_gaerdbms
2 - 注册自定义方言(可以覆盖现有模式)
from sqlalchemy.dialects import registry
registry.register("mysql.gaerdbms", "application.database.gaerdbms_dialect", "MySQLDialect_gaerdbms")
注意:0.8版本允许在当前进程中注册方言(如上所示)。如果您正在运行旧版本的SQLAlchemy,建议升级到0.8+,否则您需要根据
此处的说明为方言创建单独的安装程序。
3 - 更新create_engine('...')
URL,因为项目和实例现在作为URL的查询字符串的一部分提供。
mysql+gaerdbms:///<dbname>?instance=<project:instance>
例如:
举个例子:
create_engine('mysql+gaerdbms:///mydb?instance=myproject:instance1')