lsm

为SQLite的LSM键/值存储提供快速Python绑定。LSM存储引擎最初是作为实验性的SQLite4重写（现在已废弃）的一部分编写的。最近，LSM源代码已移动到SQLite3 源代码树，并经历了一些改进和修复。此项目使用SQLite3源代码树中的LSM代码。

功能

• 嵌入式的零配置数据库。
• 键支持使用游标进行有序遍历。
• 事务性（包括嵌套事务）。
• 基于单写多读MVCC的事务并发模型。
• 磁盘数据库存储在单个文件中。
• 在应用或电源故障的情况下，数据是持久的。
• 线程安全。
• 为读写操作释放GIL（每个连接都有自己的互斥锁）
• 页面压缩（lz4或zstd）
• 零依赖静态库
• Python 3.x。

限制

Python lsm 的源代码托管在 GitHub 上。

如果您在库中遇到任何错误，请 创建一个问题，包括错误描述和相关跟踪信息。

快速入门

以下是一个示例交互式控制台会话，旨在展示 lsm Python 库的一些基本功能和功能。

首先，实例化一个 LSM 对象，指定数据库文件的路径。

from lsm import LSM
db = LSM('test.ldb')
assert db.open()

更 Pythonic 的变体是使用上下文管理器

from lsm import LSM
with LSM("test.ldb") as db:
    assert db.info()

未打开的数据库将引发 RuntimeError

import pytest
from lsm import LSM

db = LSM('test.ldb')

with pytest.raises(RuntimeError):
    db.info()

二进制/字符串模式

您应该使用 binary: bool = True 参数选择以何种模式打开数据库。

例如，当您想存储字符串时，只需传递 binary=False

from lsm import LSM
with LSM("test_0.ldb", binary=False) as db:
    # must be str for keys and values
    db['foo'] = 'bar'
    assert db['foo'] == "bar"

否则，您必须将键和值作为 bytes 传递（默认行为）

from lsm import LSM

with LSM("test.ldb") as db:
    db[b'foo'] = b'bar'
    assert db[b'foo'] == b'bar'

键/值特性

lsm 是一个键/值存储，具有类似字典的 API

from lsm import LSM
with LSM("test.ldb", binary=False) as db:
    db['foo'] = 'bar'
    assert db['foo'] == 'bar'

数据库尽快应用更改

import pytest
from lsm import LSM

with LSM("test.ldb", binary=False) as db:
    for i in range(4):
         db[f'k{i}'] = str(i)

    assert 'k3' in db
    assert 'k4' not in db
    del db['k3']

    with pytest.raises(KeyError):
        print(db['k3'])

默认情况下，当您尝试查找键时，lsm 将搜索精确匹配。如果您要查找的特定键不存在，您也可以搜索最接近的键

import pytest
from lsm import LSM, SEEK_LE, SEEK_GE, SEEK_LEFAST


with LSM("test.ldb", binary=False) as db:
    for i in range(4):
        db[f'k{i}'] = str(i)

    # Here we will match "k1".
    assert db['k1xx', SEEK_LE] == '1'

    # Here we will match "k1" but do not fetch a value
    # In this case the value will always be ``True`` or there will
    # be an exception if the key is not found
    assert db['k1xx', SEEK_LEFAST] is True

    with pytest.raises(KeyError):
        print(db['000', SEEK_LEFAST])

    # Here we will match "k2".
    assert db['k1xx', SEEK_GE] == "2"

LSM 支持其他常见的字典方法，如

• keys()
• values()
• items()
• update()

切片和迭代

可以直接迭代数据库，也可以切片。当您正在切片数据库时，起始键和结束键不必存在 - lsm 将找到最接近的键（详细信息请参阅 LSM.fetch_range() 文档）。

from lsm import LSM

with LSM("test_slices.ldb", binary=False) as db:

    # clean database
    for key in db.keys():
        del db[key]

    db['foo'] = 'bar'

    for i in range(3):
        db[f'k{i}'] = str(i)

    # Can easily iterate over the database items
    assert (
        sorted(item for item in db.items()) == [
            ('foo', 'bar'), ('k0', '0'), ('k1', '1'), ('k2', '2')
        ]
    )

    # However, you will not read the entire database into memory, as special
    # iterator objects are used.
    assert str(db['k0':'k99']).startswith("<lsm_slice object at")

    # But you can cast it to the list for example
    assert list(db['k0':'k99']) == [('k0', '0'), ('k1', '1'), ('k2', '2')]

您可以使用开放切片。如果下限或上限超出键的范围，则返回空列表。

with LSM("test_slices.ldb", binary=False, readonly=True) as db:
    assert list(db['k0':]) == [('k0', '0'), ('k1', '1'), ('k2', '2')]
    assert list(db[:'k1']) == [('foo', 'bar'), ('k0', '0'), ('k1', '1')]
    assert list(db[:'aaa']) == []

要按逆序检索键或跳过多个项，只需使用第三个切片参数即可。负步值表示逆序，但第一和第二个参数必须是常规顺序。

with LSM("test_slices.ldb", binary=False, readonly=True) as db:
    assert list(db['k0':'k99':2]) == [('k0', '0'), ('k2', '2')]
    assert list(db['k0'::-1]) == [('k2', '2'), ('k1', '1'), ('k0', '0')]
    assert list(db['k0'::-2]) == [('k2', '2'), ('k0', '0')]
    assert list(db['k0'::3]) == [('k0', '0')]

您还可以 删除 键的切片，但请注意，删除 不会 包括键本身

with LSM("test_slices.ldb", binary=False) as db:
    del db['k0':'k99']

    # Note that 'k0' still exists.
    assert list(db.items()) == [('foo', 'bar'), ('k0', '0')]

游标

虽然切片可以覆盖大多数用例，但对于更精细的控制，您可以使用游标遍历记录。

from lsm import LSM, SEEK_GE, SEEK_LE

with LSM("test_cursors.ldb", binary=False) as db:
    del db["a":"z"]

    db["spam"] = "spam"

    with db.cursor() as cursor:
        cursor.seek('spam')
        key, value = cursor.retrieve()
        assert key == 'spam'
        assert value == 'spam'

游标查找

with LSM("test_cursors.ldb", binary=False) as db:
    db.update({'k0': '0', 'k1': '1', 'k2': '2', 'k3': '3', 'foo': 'bar'})

    with db.cursor() as cursor:

        cursor.first()
        key, value = cursor.retrieve()
        assert key == "foo"
        assert value == "bar"

        cursor.last()
        key, value = cursor.retrieve()
        assert key == "spam"
        assert value == "spam"

        cursor.previous()
        key, value = cursor.retrieve()
        assert key == "k3"
        assert value == "3"

查找大于或等于 'k0' 的第一个匹配项，然后向前移动直到键小于 'k99'

with LSM("test_cursors.ldb", binary=False) as db:
    with db.cursor() as cursor:
        cursor.seek("k0", SEEK_GE)
        results = []

        while cursor.compare("k99") > 0:
            key, value = cursor.retrieve()
            results.append((key, value))
            cursor.next()

    assert results == [('k0', '0'), ('k1', '1'), ('k2', '2'), ('k3', '3')]

查找小于或等于 'k99' 的最后一个匹配项，然后向后移动直到键小于 'k0'

with LSM("test_cursors.ldb", binary=False) as db:
    with db.cursor() as cursor:
        cursor.seek("k99", SEEK_LE)
        results = []

        while cursor.compare("k0") >= 0:
            key, value = cursor.retrieve()
            results.append((key, value))
            cursor.previous()

    assert results == [('k3', '3'), ('k2', '2'), ('k1', '1'), ('k0', '0')]

当您使用完毕时，关闭游标非常重要。因此，建议您使用 LSM.cursor() 上下文管理器，以确保游标正确关闭。

事务

lsm 支持嵌套事务。使用事务的最简单方法是使用 LSM.transaction() 方法，该方法返回一个上下文管理器

from lsm import LSM

with LSM("test_tx.ldb", binary=False) as db:
    del db["a":"z"]
    for i in range(10):
        db[f"k{i}"] = f"{i}"


with LSM("test_tx.ldb", binary=False) as db:
    with db.transaction() as tx1:
        db['k1'] = '1-mod'

        with db.transaction() as tx2:
            db['k2'] = '2-mod'
            tx2.rollback()

    assert db['k1'] == '1-mod'
    assert db['k2'] == '2'

您可以在包装块中部分提交或回滚事务

from lsm import LSM

with LSM("test_tx_2.ldb", binary=False) as db:
    del db["a":"z"]
    for i in range(10):
        db[f"k{i}"] = f"{i}"

with LSM("test_tx_2.ldb", binary=False) as db:
    with db.transaction() as txn:
        db['k1'] = 'outer txn'

        # The write operation is preserved.
        txn.commit()

        db['k1'] = 'outer txn-2'

        with db.transaction() as txn2:
            # This is committed after the block ends.
            db['k1'] = 'inner-txn'

        assert db['k1'] == "inner-txn"

        # Rolls back both the changes from txn2 and the preceding write.
        txn.rollback()

        assert db['k1'] == 'outer txn', db['k1']

如果您愿意，也可以显式调用 LSM.begin()、LSM.commit() 和 LSM.rollback()。

from lsm import LSM

# fill db
with LSM("test_db_tx.ldb", binary=False) as db:
    del db["k":"z"]
    for i in range(10):
        db[f"k{i}"] = f"{i}"


with LSM("test_db_tx.ldb", binary=False) as db:
    # start transaction
    db.begin()
    db['k1'] = '1-mod'

    # nested transaction
    db.begin()
    db['k2'] = '2-mod'
    # rolling back nested transaction
    db.rollback()

    # comitting top-level transaction
    db.commit()

    assert db['k1'] == '1-mod'
    assert db['k2'] == '2'

感谢

• @coleifer - 此项目受 coleifer/python-lsm-db 的启发。