如何在Python中将一个内存中的SQLite数据库复制到另一个内存数据库？

Question

我正在为Django写一个测试套件，这个测试会像树一样运行。比如，测试案例A可能有2种结果，测试案例B可能有1种结果，而测试案例C可能有3种结果。这个树的结构看起来是这样的：

      X
     /
A-B-C-X
 \   \
  B   X
   \   X
    \ /
     C-X
      \
       X

在上面这个树的每一条路径上，数据库的内容可能会不同。所以在每个分叉点，我想创建一个当前数据库状态的内存副本，然后把这个副本作为参数传递给下一个测试。

有没有人知道怎么把内存中的数据库复制到另一个数据库，然后再获取一个引用来传递这个数据库呢？

谢谢！

Answer 1

这里有一个函数，用来复制数据库。源数据库和目标数据库可以是在内存中，也可以是在硬盘上（默认情况下，目标是复制到内存中）：

import sqlite3

def copy_database(source_connection, dest_dbname=':memory:'):
    '''Return a connection to a new copy of an existing database.                        
       Raises an sqlite3.OperationalError if the destination already exists.             
    '''
    script = ''.join(source_connection.iterdump())
    dest_conn = sqlite3.connect(dest_dbname)
    dest_conn.executescript(script)
    return dest_conn

下面是一个如何将这个函数应用到你的具体情况的例子：

from contextlib import closing

with closing(sqlite3.connect('root_physical.db')) as on_disk_start:
    in_mem_start = copy_database(on_disk_start)

a1 = testcase_a_outcome1(copy_database(in_mem_start))
a2 = testcase_a_outcome1(copy_database(in_mem_start))
a1b = test_case_b(a1)
a2b = test_case_b(a2)
a1bc1 = test_case_c_outcome1(copy_database(a1b))
a1bc2 = test_case_c_outcome2(copy_database(a1b))
a1bc3 = test_case_c_outcome3(copy_database(a1b))
a2bc1 = test_case_c_outcome1(copy_database(a2b))
a2bc2 = test_case_c_outcome2(copy_database(a2b))
a2bc3 = test_case_c_outcome3(copy_database(a2b))

Answer 2

好的，经过一番有趣的探索，我搞定了这个问题。

from django.db import connections
import sqlite3

# Create a Django database connection for our test database
connections.databases['test'] = {'NAME': ":memory:", 'ENGINE': "django.db.backends.sqlite3"}

# We assume that the database under the source_wrapper hasn't been created
source_wrapper = connections['default']  # put alias of source db here
target_wrapper = connections['test'] 

# Create the tables for the source database
source_wrapper.creation.create_test_db()

# Dump the database into a single text query
query = "".join(line for line in source_wrapper.connection.iterdump())

# Generate an in-memory sqlite connection
target_wrapper.connection = sqlite3.connect(":memory:")
target_wrapper.connection.executescript(query)

现在，名为 test 的数据库将和 default 数据库一模一样。你可以用 target_wrapper.connection 来引用新创建的数据库。