Python package for OrjsonTranscoder.
Project description
Welcome to the OrjsonTranscoder project
This package provides a OrjsonTranscoder
class for use with
the Python eventsourcing library that uses the orjson
library.
Installation
Use pip to install the stable distribution from the Python Package Index.
$ pip install eventsourcing_orjsontranscoder
It is recommended to install Python packages into a Python virtual environment.
This package uses Cython, so relevant build tools may need to be installed before this package can be installed successfully.
Synopsis
>>> 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)
Features
Most importantly, OrjsonTranscoder
supports custom transcoding of instances
of tuple
and subclasses of str
, int
, dict
and tuple
. This is an
important improvement on the core library's JSONTranscoder
class which converts
tuple
to list
and loses type information for subclasses of str
, int
, dict
and tuple
.
It is also faster than JSONTranscoder
, encoding approximately x3 faster
and decoding approximately x2 faster. This is less important than the preservation
of type information (see above) because latency in your application will
usually be dominated by database interactions. However, it's nice that it
is not slower.
class | encode | decode |
---|---|---|
OrjsonTranscoder | 6.8 μs | 13.8 μs |
JSON Transcoder | 20.1 μs | 25.7 μs |
The above benchmark was performed on GitHub using the following object, which is perhaps representative of the state of a domain event in an event-sourced application.
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"))
),
}
Custom Transcodings
Define custom transcodings for your custom value object types by subclassing
CTranscoding
. The prefix C
is used to distinguish these classes from the
Transcoding
classes provided by the core Python eventsourcing library.
For example, consider the custom value object MyInt
below.
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)
You can define a custom transcoding for MyInt
as a normal Python class in a
normal Python module (.py
file) using the CTranscoding
class.
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)
Alternatively for greater speed, you can define a custom transcoding for MyInt
as a Cython extension type class in a Cython module (.pyx
file) using the
CTranscoding
extension type. See this project's Git repository for examples.
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)
If you define Cython modules, you will need to build them in-place before you can use them. If you are distributing your code, you will also need to configure your distribution to build the Cython module when your code is installed.
$ cythonize -i my_transcodings.pyx
See the Cython documentation for more information about Cython.
Using the OrjsonTranscoder
To use the OrjsonTranscoder
class in a Python eventsourcing application
object, override the construct_transcoder()
and register_transcodings()
methods.
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
See the library docs
for more information about transcoding, but please note the CTranscoder
uses a slightly
different API.
Developers
After cloning the repository, you can set up a virtual environment and install dependencies by running the following command in the root folder.
$ make install
After making changes, please run the tests.
$ make test
Check the formatting of the code.
$ make lint
You can automatically reformat the code by running the following command.
$ make fmt
If the project dependencies change, you can update your packages by running the following command.
$ make update-packages
Please submit changes for review by making a pull request.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Hashes for eventsourcing_orjsontranscoder-0.1.2.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | d4ecabfcbc1ef716b39cdfc047f87ba11216dcf597d88c0a7f5e6599a0f31457 |
|
MD5 | e31ba6c27560ef513b0c73595975ef57 |
|
BLAKE2b-256 | f8a6a32754e2ef30b4472e267bcccfc1cf846bf6165c49206ed62ed1cd97b2cb |