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

【问题标题】：How can I copy an in-memory SQLite database to another in-memory SQLite database in Python?如何在 Python 中将内存中的 SQLite 数据库复制到另一个内存中的 SQLite 数据库？
【发布时间】：2011-11-08 03:27:43
【问题描述】：

我正在为 Django 编写一个测试套件，它以树状方式运行测试。例如，Testcase A 可能有 2 个结果，Testcase B 可能有 1 个结果，Testcase C 可能有 3 个结果。树看起来像这样

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

对于上面树中的每个路径，数据库内容可能不同。因此，在每次分叉时，我都在考虑创建数据库当前状态的内存副本，然后将该参数输入到下一个测试中。

有人知道如何从本质上将内存数据库复制到另一个数据库，然后获取引用以传递该数据库吗？

谢谢！

【问题讨论】：

什么是内存数据库（在 django 或 python 方面）？
一个未存储在磁盘上。在 Django 中，您可以通过指定 None 或 :memory: 作为数据库名称来指定 SQLite 使用内存数据库。
谢谢。这意味着您的问题是如何从同一个应用程序复制 django 应用程序正在使用的数据库，对吗？

标签： python django sqlite

【解决方案1】：

好吧，经过一次有趣的冒险，我想出了这个。

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 作为对新创建数据库的引用。

【讨论】：

我知道已经五年半了，但是伙计，你让我很开心！我找不到将我的测试内存数据库转储到文件的方法，现在它完成了。如果您不介意，我也冒昧地对您的答案进行了一些改进。

【解决方案2】：

这是一个复制数据库的函数。源和目标都可以是内存或磁盘（默认目标是内存中的副本）：

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))

【讨论】：