3

我从我的开发工作站使用 SQLAlchemy 0.7.9 连接到 Google Cloud SQL 失败（希望使用 create_all() 生成架构）。我无法通过以下错误：

sqlalchemy.exc.DBAPIError: (AssertionError) No api proxy found for service "rdbms" None None

我能够使用 成功连接到数据库实例google_sql.py instancename，它最初打开了一个浏览器来授权连接（现在似乎已经缓存了授权，尽管我没有https://developers.google~/Library/Preferences/com.google.cloud.plist的文件。 com/cloud-sql/docs/commandline表明我应该）

这是我用来测试连接的简单应用程序：

from sqlalchemy import create_engine

engine = create_engine('mysql+gaerdbms:///myapp', connect_args={"instance":"test"})
connection = engine.connect()

完整的堆栈跟踪可在此处获得 - https://gist.github.com/4486641

4

2 回答 2

4

事实证明mysql+gaerdbms:///，SQLAlchemy 中的驱动程序仅设置为使用rdbms_apiproxyDBAPI，只能在从 Google App Engine 实例访问 Google Cloud SQL 时使用。我向 SQLAlchemy 提交了一张票以更新驱动程序，以便rdbms_googleapi在不在 Google App Engine 上时使用基于 OAuth 的驱动程序，就像App Engine SDK 中提供的Django 驱动程序一样。rdbms_googleapi 也是使用的 DBAPI google_sql.py（远程 sql 控制台）。

更新后的方言预计将成为 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):
        # Cloud SQL connections die at any moment
        return NullPool

    def create_connect_args(self, url):
        opts = url.translate_connect_args()
        opts['dsn'] = ''  # unused but required to pass to rdbms.connect()
        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')
于 2013-01-11T21:44:40.673 回答
3

我想我有一个工作示例脚本如下。你能尝试类似的事情，让我知道它是怎么回事吗？

但是（您可能已经意识到了），如果您的目标是在 Google Cloud SQL 实例上创建架构，也许您可​​以针对本地 mysql 服务器创建架构，使用 mysqldump 转储它，然后将架构导入 Google云 SQL。无论如何，使用本地 mysql 服务器进行开发非常方便。

from sqlalchemy import create_engine
from google.storage.speckle.python.api import rdbms as dbi_driver
from google.storage.speckle.python.api import rdbms_googleapi

INSTANCE = 'YOURPROJECT:YOURINSTANCE'
DATABASE = 'YOURDATABASE'

def get_connection():       
    return rdbms_googleapi.connect('', instance=INSTANCE, database=DATABASE)

def main():
    engine = create_engine(
        'mysql:///YOURPROJECT:YOURINSTANCE/YOURDATABASE',
        module=dbi_driver,
        creator=get_connection,
    )
    print engine.execute('SELECT 1').scalar()

if __name__ == '__main__':
    main()
于 2013-01-10T07:22:07.200 回答