如何在 Sqlite 数据库中将 save/restore/add 元素设置为 Python 集？

Question

如何在 Sqlite3 数据库中保存（以及还原、添加元素）一组字符串？

这不起作用，因为集不是 JSON-可序列化的：

import sqlite3, json
db = sqlite3.connect(':memory:')
db.execute('CREATE TABLE t(id TEXT, myset TEXT);')
s = {'a', 'b', 'c'}
db.execute("INSERT INTO t VALUES (?, ?);", ('1', json.dumps(s)))  
# Error: Object of type set is not JSON serializable

因此我们可以使用 list，或带有虚拟值的字典：

s = list(s)
# or s = {'a':0, 'b':0, 'c': 0}
db.execute("INSERT INTO t VALUES (?, ?);", ('1', json.dumps(s)))  

# RETRIEVE A SET FROM DB
r = db.execute("SELECT myset FROM t WHERE id = '1'").fetchone()
if r is not None:
    s = set(json.loads(r[0]))
    print(s)

然后将字符串元素添加到数据库中已有的集合中不是很优雅：

必须 SELECT,
检索为字符串，
用json.loads、
从列表转换为集合，
向集合中添加一个元素，
从集合转换为列表（或者，作为最后 3 个步骤的替代方法：检查元素是否已存在于列表中，然后将其添加或不添加到列表中）
JSON用 json.dumps,
数据库UPDATE

在 Sqlite 数据库中是否有更 pythonic 的方式来处理集合？

Answer 1

将集合存储在 SQLite 数据库中的一种更简单的方法是对 myset 列使用 BLOB 数据类型，并使用 pickle.dumps:[=16= 将其序列化为字节]

import sqlite3
import pickle

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE t(id TEXT, myset BLOB)")

s = {"a", "b", "c"}
db.execute(
    "INSERT INTO t VALUES (?, ?)",
    ("1", sqlite3.Binary(pickle.dumps(s, pickle.HIGHEST_PROTOCOL))),
)

r = db.execute("SELECT myset FROM t WHERE id = '1'").fetchone()
if r is not None:
    s = pickle.loads(r[0])
    print(s)

要向数据库中的集合添加新元素，仍然需要序列化和反序列化步骤，但不再需要转换to/from列表或检查集合中已经存在的元素。

或者，您可以确保 ID-element 组合在 database-level 处的唯一性，例如使用复合主键：

import sqlite3

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE t(id TEXT, element TEXT, PRIMARY KEY(id, element))")

s = {"a", "b", "c"}
for element in s:
    db.execute("INSERT INTO t VALUES(?, ?)", ("1", element))

r = db.execute("SELECT element FROM t WHERE id = '1'").fetchall()
s = set(row[0] for row in r)
print(s)

Answer 2

您可以使用 sqlite 注册 adapter and converter 函数，它将自动执行所需的转换。

import json
import sqlite3

def adapt_set(value):
    return json.dumps(list(value))

def convert_set(value):
    return set(json.loads(value))

sqlite3.register_adapter(set, adapt_set)
sqlite3.register_converter('set_type', convert_set)

注册这些函数后，将detect_types传递给连接工厂以告诉sqlite如何使用它们。

传递 sqlite3.PARSE_DECLTYPE 将使连接使用声明的类型来查找 adapter/converter。

db = sqlite3.connect(':memory:', detect_types=sqlite3.PARSE_DECLTYPES)

# Declare the myset column as type "set_type". 
db.execute('CREATE TABLE t(id TEXT, myset set_type);')

db.execute("INSERT INTO t VALUES (?, ?);", ('1', {1, 2, 3})) 
r = db.execute("""SELECT myset FROM t WHERE id = '1'""").fetchone()
print(r[0])   # <- r[0] is a set.

传递 sqlite.PARSE_COLNAMES 将导致在游标描述中的列名称中搜索方括号中的类型名称。

db = sqlite3.connect(':memory:', detect_types=sqlite3.PARSE_COLNAMES)
# The type is not declared in the created table statement.
db.execute('CREATE TABLE t(id TEXT, myset TEXT);')
db.execute("INSERT INTO t VALUES (?, ?);", ('1', {1, 2, 3})) 

# Include the type in the column label.
r = db.execute("""SELECT myset "AS myset [set_type]" FROM t WHERE id = '1'""").fetchone()
print(r[0])   <- r[0] is a set

Answer 3

我会注册一个“集合适配器”，通过简单地获取集合的字符串表示并将其编码为字节字符串进行存储，将集合转换为字节字符串，并注册一个“集合转换器”，将我们的 user-defined 列类型命名为“set_type”（选择您想要的任何替代项），通过将字节字符串解码回 Unicode 字符串然后对其应用 eval，将字节字符串返回到集合中：

import sqlite3
from decimal import Decimal


db = sqlite3.connect(':memory:', detect_types=sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES)

def set_adapter(the_set):
    return str(the_set).encode('utf-8')

def set_converter(s):
    return eval(s.decode('utf-8'))

sqlite3.register_adapter(set, set_adapter)
sqlite3.register_converter('set_type', set_converter)


# Define the columns with type set_type:
db.execute('CREATE TABLE t(id TEXT, myset set_type);')
s = {'a', 'b', 'c', (1, 2, 3), True, False, None, b'abcde', Decimal('12.345')}
# Our adapter will store s as a byte string:
db.execute("INSERT INTO t VALUES (?, ?);", ('1', s))
cursor = db.cursor()
# Our converter will convert column type set_type from bytes to set:
cursor.execute('select myset from t')
row = cursor.fetchone()
s = row[0]
print(type(s), s)
db.close()

打印：

<class 'set'> {False, True, 'b', None, (1, 2, 3), Decimal('12.345'), b'abcde', 'a', 'c'}

如您所见，在将集合转换为列表后，这还可以处理比 JSON 更多的数据类型。 JSON 可能会处理尽可能多的数据类型，但前提是您编写 JSON 转换器（这通常意味着将这些数据类型转换为支持的类型之一，例如字符串）。

如 Jonathan Feenstra 提供的答案那样定义适配器和转换器以使用 pickle 将导致速度提高，但可能会使用更多存储空间。但我会使用上面概述的技术，即使用具有特殊 user-defined 列类型的适配器和转换器函数：

import pickle

def set_adapter(the_set):
    return pickle.dumps(the_set, pickle.HIGHEST_PROTOCOL)

def set_converter(s):
    return pickle.loads(s)

Answer 4

你可以使用aiosqlitedict

这是它的功能

Easy conversion between sqlite table and Python dictionary and vice-versa.

Get values of a certain column in a Python list.

Order your list ascending or descending.

Insert any number of columns to your dict.

入门

我们首先连接我们的数据库参考栏目

from aiosqlitedict.database import Connect

countriesDB = Connect("database.db", "user_id")

制作字典

字典应该在异步函数中。

async def some_func():
    countries_data = await countriesDB.to_dict("my_table_name", 123, "col1_name", "col2_name", ...)

您可以插入任意数量的列，或者您可以通过指定获取全部列名称为“*”

    countries_data = await countriesDB.to_dict("my_table_name", 123, "*")

所以你现在已经对你的词典做了一些修改并且想要再次将其导出为 sql 格式？

将 dict 转换为 sqlite table

async def some_func():
    ...
    await countriesDB.to_sql("my_table_name", 123, countries_data)

但是如果您想要特定列的值列表怎么办？

Select方法

您可以获得特定列的所有值的列表。

country_names = await countriesDB.select("my_table_name", "col1_name")

使用 limit 参数限制您的选择。

country_names = await countriesDB.select("my_table_name", "col1_name", limit=10)

您还可以使用 ascending 参数来安排您的 list and/or order_by 参数并指定特定列以相应地对列表进行排序。

country_names = await countriesDB.select("my_table_name", "col1_name", order_by="col2_name", ascending=False)

如何在 Sqlite 数据库中将 save/restore/add 元素设置为 Python 集？

How to save/restore/add elements to Python sets in a Sqlite database?

python

sqlite

serialization

json

set

入门

制作字典

将 dict 转换为 sqlite table

Select方法