python - 如何使用 Python 的 DBAPI 来可移植地验证数据库模式？

Question

我正在编写一个实用程序，我打算将其与至少三个不同的数据库服务器后端(SQLite3、PostgreSQL 和 MySQL)一起使用。我没有使用 ORM(尽管在完成原型(prototype)并熟悉如何维护架构以便其他工具可能访问(将来可能来自其他语言)后，我尝试使用 SQLAlchemy 进行 fork )。

我发布此内容是本着 StackOverflow 关于回答自己的问题以及开放问题和答案进行讨论的政策的精神。

下面的答案中发布的代码已经针对所有三个受支持的目标进行了测试，这些目标预先配置了我的预期架构的一个非常简单的子集(靠近脚本顶部的硬编码)。目前它仅检查每个表是否具有所有必需的列，并且仅在存在额外列时发出警告。我不检查列内容的类型或其他约束。

(我的实用程序应该允许最终用户添加其他列，只要他们允许 NULL 值或代码提供合适的默认值 --- 因为我的代码只会插入或更新我所使用的列的子集我正在指定)。

如果有更好的方法来完成此任务，或者我是否犯了一些在我的测试中未发现的严重错误，请告诉我。

(我想知道如何正确创建SQLFiddle条目/链接，可以共享这些条目/链接以使这个问题更容易处理。欢迎任何关于这一点的指示，我已经尝试在这里创建这个my MySQL schema ...让我们看看这是否有效)。

Answer 1

如上所述，这是我的方法:

#!/usr/bin/env python
import importlib
import sys

schema = (('tags',     set(('id','tag','gen','creation'))),
          ('nodes',    set(('name', 'id', 'gen', 'creation'))),
          ('node_tag', set(('node_id', 'tag_id', 'creation'))),
          ('tgen',     set(('tid', 'comment', 'creation')))
         )

drivers = {
           'mysql':  'MySQLdb',
           'pg':     'psycopg2',
           'sqlite': 'sqlite3',
          }

if __name__ == '__main__':
    args = sys.argv[1:]
    if args[0] in drivers.keys():
        dbtype = args[0]
        db = importlib.import_module(drivers[dbtype])
    else:
        print >> sys.stderr, 'Unrecognized dbtype %s, should be one of these:' % args[0], ' '.join(drivers.keys())
        sys.exit(127)

    if dbtype == 'sqlite':
        required_args = 2
        dbopts = { 'database': args[1] }
    else:
        required_args = 6
        dbopts = { 'database' : args[1],
                   'user'     : args[2],
                   'passwd'   : args[3],
                   'host'     : args[4],
                   'port'     : int(args[5]),
                }

if len(args) < required_args:
    print >> sys.stderr, 'Must supply all arguments:',
    print >> sys.stderr, '[mysql|pg|sqlite] database user passwd host port'
    sys.exit(126)

if dbtype == 'mysql':
    dbopts['db'] = dbopts['database']
    del dbopts['database']

if dbtype == 'pg':
    dbopts['password'] = dbopts['passwd']
    del dbopts['passwd']

try:
    conn = db.connect(**dbopts)
except db.Error, e:
    print 'Database connection failed: %s' % e
    sys.exit(125)

cur  = conn.cursor()

exit_code = 0
for each_table in schema:
    table, columns = each_table
    introspected_columns = None
    try:
        cur.execute("SELECT * FROM %s WHERE 0=1" % table)
        introspected_columns = set((x[0] for x in cur.description))
    except db.Error, e:
        print >> sys.stderr, 'Encountered %s Error on table %s' % (e, table)
    if introspected_columns:
        missing = columns - introspected_columns
        extra   = introspected_columns - columns
    if missing:
        print 'Error: Missing columns in table %s: %s' % (table,' '.join(missing))
        exit_code += 1
    else:
        print 'All columns present for table %s' % table
    if extra:
        print 'Warning: Extra columns in table %s: %s' % (table, ' '.join(extra))

sys.exit(exit_code)

...在实际实践中，这将被重构为一个类，并在我正在编写的守护进程的服务启动期间调用。

请注意，这里的实际技术是对所有列进行查询 SELECT * FROM some_table ...但保证不返回任何行 WHERE 0=1 ...这意味着我不需要提前知道任何表名。这似乎一致地设置了 DBAPI(客户端)游标中的 DBAPI cur.description 字段。

(顺便说一句，这个示例还强调了流行的 DBAPI 数据库驱动程序中连接关键字参数的一些恼人的差异。它们似乎都没有忽略额外的键，因此我们需要为这三个驱动程序中的每一个设置明显不同的参数集；只需 SQLite3 的文件名，并将 MySQLdb 的“database”键名称更改为“db”，将 PostgreSQL 的“password”更改为“passwd”——至少对于 psycopg2 驱动程序而言)。