欢迎来到OrjsonTranscoder项目

此包提供orjson库的OrjsonTranscoder类，用于与Python事件源库一起使用。

安装

使用pip从Python包索引安装稳定发行版。

$ pip install eventsourcing_orjsontranscoder

建议将Python包安装到Python虚拟环境中。

此包使用Cython，因此在成功安装此包之前可能需要安装相关的构建工具。

概要

>>> from eventsourcing_orjsontranscoder import OrjsonTranscoder, CTupleAsList
>>> t = OrjsonTranscoder()
>>> t.register(CTupleAsList())
>>> d = t.encode((1,2,3))
>>> d
b'{"_type_":"tuple_as_list","_data_":[1,2,3]}'
>>> t.decode(d)
(1, 2, 3)

功能

最重要的是，OrjsonTranscoder支持对tuple实例和str、int、dict和tuple子类的自定义转码。这是对核心库的JSONTranscoder类的一个重要改进，该类将tuple转换为list，并丢失了str、int、dict和tuple子类的类型信息。

它也比JSONTranscoder更快，编码速度大约是x3，解码速度大约是x2。虽然与保留类型信息（见上文）相比这不太重要，因为应用程序中的延迟通常会被数据库交互主导。然而，它不慢是件好事。

上述基准测试是在GitHub上使用以下对象进行的，该对象可能代表了事件源应用程序中领域事件的当前状态。

obj = {
    "originator_id": uuid5(NAMESPACE_URL, "some_id"),
    "originator_version": 123,
    "timestamp": DomainEvent.create_timestamp(),
    "a_str": "hello",
    "b_int": 1234567,
    "c_tuple": (1, 2, 3, 4, 5, 6, 7),
    "d_list": [1, 2, 3, 4, 5, 6, 7],
    "e_dict": {"a": 1, "b": 2, "c": 3},
    "f_valueobj": CustomType2(
        CustomType1(UUID("b2723fe2c01a40d2875ea3aac6a09ff5"))
    ),
}

自定义转码

通过继承 CTranscoding 来为自定义值对象类型定义自定义转码。前缀 C 用于区分这些类与由核心 Python eventsourcing 库提供的 Transcoding 类。

例如，考虑以下自定义值对象 MyInt。

class MyInt(int):
    def __repr__(self):
        return f"{type(self).__name__}({super().__repr__()})"

    def __eq__(self, other):
        return type(self) == type(other) and super().__eq__(other)

您可以使用 CTranscoding 类在常规 Python 模块（.py 文件）中定义 MyInt 的自定义转码，作为常规 Python 类。

class CMyIntAsInt(CTranscoding):
    def type(self):
        return MyInt

    def name(self):
        return "myint_as_int"

    def encode(self, obj):
        return int(obj)

    def decode(self, data):
        return MyInt(data)

或者为了更高的速度，您可以使用 CTranscoding 扩展类型在 Cython 模块（.pyx 文件）中定义 MyInt 的自定义转码，作为 Cython 扩展类型类。请参阅此项目的 Git 仓库以获取示例。

from _eventsourcing_orjsontranscoder cimport CTranscoding

from my_domain_model import MyInt

cdef class CMyIntAsInt(CTranscoding):
    cpdef object type(self):
        return MyInt

    cpdef object name(self):
        return "myint_as_int"

    cpdef object encode(self, object obj):
        return int(obj)

    cpdef object decode(self, object data):
        return MyInt(data)

如果您定义 Cython 模块，您需要在使用之前在原地构建它们。如果您正在分发代码，您还需要配置您的分发以在代码安装时构建 Cython 模块。

$ cythonize -i my_transcodings.pyx

有关 Cython 的更多信息，请参阅 Cython 文档。

使用 OrjsonTranscoder

要在 Python eventsourcing 应用程序对象中使用 OrjsonTranscoder 类，请重写 construct_transcoder() 和 register_transcodings() 方法。

from eventsourcing.application import Application
from eventsourcing.domain import Aggregate, event
from eventsourcing_orjsontranscoder import (
    CDatetimeAsISO,
    CTupleAsList,
    CUUIDAsHex,
    OrjsonTranscoder,
)


class DogSchool(Application):
    def construct_transcoder(self):
        transcoder = OrjsonTranscoder()
        self.register_transcodings(transcoder)
        return transcoder

    def register_transcodings(self, transcoder):
        transcoder.register(CDatetimeAsISO())
        transcoder.register(CTupleAsList())
        transcoder.register(CUUIDAsHex())
        transcoder.register(CMyIntAsInt())

    def register_dog(self, name, age):
        dog = Dog(name, age)
        self.save(dog)
        return dog.id

    def add_trick(self, dog_id, trick):
        dog = self.repository.get(dog_id)
        dog.add_trick(trick)
        self.save(dog)

    def update_age(self, dog_id, age):
        dog = self.repository.get(dog_id)
        dog.update_age(age)
        self.save(dog)

    def get_dog(self, dog_id):
        dog = self.repository.get(dog_id)
        return {"name": dog.name, "tricks": tuple(dog.tricks), "age": dog.age}


class Dog(Aggregate):
    @event("Registered")
    def __init__(self, name, age):
        self.name = name
        self.age = age
        self.tricks = []

    @event("TrickAdded")
    def add_trick(self, trick):
        self.tricks.append(trick)

    @event("AgeUpdated")
    def update_age(self, age):
        self.age = age


def test_dog_school():
    # Construct application object.
    school = DogSchool()

    # Evolve application state.
    dog_id = school.register_dog("Fido", MyInt(2))
    school.add_trick(dog_id, "roll over")
    school.add_trick(dog_id, "play dead")
    school.update_age(dog_id, MyInt(5))

    # Query application state.
    dog = school.get_dog(dog_id)
    assert dog["name"] == "Fido"
    assert dog["tricks"] == ("roll over", "play dead")
    assert dog["age"] == MyInt(5)

    # Select notifications.
    notifications = school.notification_log.select(start=1, limit=10)
    assert len(notifications) == 4

有关转码的更多信息，请参阅库文档，但请注意 CTranscoder 使用略有不同的 API。

开发者

在克隆存储库后，您可以在根目录中运行以下命令来设置虚拟环境和安装依赖项。

$ make install

在做出更改后，请运行测试。

$ make test

检查代码格式。

$ make lint

您可以通过运行以下命令自动重新格式化代码。

$ make fmt

如果项目依赖项更改，您可以通过运行以下命令更新您的包。

$ make update-packages

请通过创建拉取请求来提交更改以供审查。