Pythonによるイベントソーシングへの挑戦と現状に対する考察 / Challenging Event Sourcing with Python and Reflections on the Current State

PyConJP 2024 のセッション用スライドです
Python eventsourcing がメインとなっています

Proposal: https://2024.pycon.jp/ja/talk/DAH78A

本トークではイベントソーシングの概念や利点をお話するのと同時に、Pythonにおけるイベントソーシングに挑戦した結果を共有します。

私は普段、JVM でイベントソーシングをベースとしたマイクロサービス群の開発に取り組んでいます。 イベントソーシングはシステム上の出来事をイベントとして永続化することでシステムを構築する手法です。 システムをイベントでとらえることは多くの利点をもたらします。

ひとつ分かりやすい例を挙げるとすれば、システムの全ての状態変化を時間を追って追跡することが可能となることが挙げられます。 たとえば、何らかの問題が発生した場合には、発生したイベントのシーケンスをたどることで問題の原因を容易に特定できます。 ちょうどGitのログをたどるのと同じイメージです。 これはシステムの運用を確実に楽にします。

その他にもイベントデータはシステム間の連携を疎結合にしますし、永続化されたイベントデータを利用して、あとからBIツールと連携するといった芸当も可能です。

本トークではそのように便利なイベントソーシングの基本原理やメリットなどをお話しし、またイベントソーシングを実践する上で諸問題への回答となる各種機能について網羅して解説します。 また、それらの前提を共有した上で、現時点で Python における実装に挑戦した結果を紹介し、実践にむけて越えなくてはいけないハードル＝「Python で現状実現できてないこと」について探ります。

# URL
YouTube: https://www.youtube.com/c/narusemi
HomePage: https://nrslib.com
Twitter: https://twitter.com/nrslib
Instagram: https://www.instagram.com/nrslib/

Transcript

94. [tool.poetry] name = "pyes-fastapi" version = "0.1.0" description = "" authors = ["Your Name <you@example.com>"] readme = "README.md" [tool.poetry.dependencies] python = "^3.10" uvicorn = {extras = ["standard"], version = "^0.24.0.post1"} fastapi = "^0.109.1" sqlalchemy = "^2.0.0" fastapi-sqlalchemy = "^0.2.1" pydantic = "^1.10.12" # latest eventsourcing = { version = "~=9.3", extras = ["sqlite"] } eventsourcing_sqlalchemy = "^0.8" # with Axon #eventsourcing = { version = ">=9.2,<9.3", extras = ["sqlite"] } #eventsourcing-axonserver="^0.1.5" #eventsourcing_sqlalchemy = "^0.7" [build-system] requires = ["poetry-core"] build-backend = "poetry.core.masonry.api"
98. @router.post("/documents/") def post_docs(request: DocsPost, doc_service: DocService = Depends(get_doc_service)): doc_id = doc_service.create(request.contents) return doc_id
101. class DocService(Application): def create(self, body): doc = Doc(body) self.save(doc) return doc.id class Doc(Aggregate): body: str class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body :
102. class DocService(Application): def create(self, body): doc = Doc(body) self.save(doc) return doc.id class Doc(Aggregate): body: str class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body :
103. class DocService(Application): def create(self, body): doc = Doc(body) self.save(doc) return doc.id class Doc(Aggregate): body: str class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body :
104. class DocService(Application): def create(self, body): doc = Doc(body) self.save(doc) return doc.id class Doc(Aggregate): body: str class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body :
105. { "timestamp": { "_type_": "datetime_iso", "_data_": "2024-09-25T10:26:16.814625+00:00" }, "originator_topic": "app.module.document.domain.models.doc:Doc", "body": "hello" } app.module.document.domain.models.doc:Doc.DocCreated topic: state:
108. system = System( pipes=[ [DocService, DocumentDataModelProjection], [EffectiveService, DocumentDataModelProjection], ] ) runner = MultiThreadedRunner # SingleThreadedRunner にすると同期的に処理される runner = runner(system, None) runner.start() doc_service = runner.get(DocService)
109. system = System( pipes=[ [DocService, DocumentDataModelProjection], [EffectiveService, DocumentDataModelProjection], ] ) runner = MultiThreadedRunner # SingleThreadedRunner にすると同期的に処理される runner = runner(system, None) runner.start() doc_service = runner.get(DocService)
110. class DocumentDataModelProjection(ProcessApplication): @policy.register(Doc.DocCreated) def handle(self, event: Doc.DocCreated, process_event): document_create = DocumentCreate( document_id=event.originator_id, body=event.body, effective_count=0 ) with get_db() as db: create_document(db, document_create)
111. class DocumentDataModelProjection(ProcessApplication): @policy.register(Doc.DocCreated) def handle(self, event: Doc.DocCreated, process_event): document_create = DocumentCreate( document_id=event.originator_id, body=event.body, effective_count=0 ) with get_db() as db: create_document(db, document_create)
112. class DocumentDataModelProjection(ProcessApplication): @policy.register(Doc.DocCreated) def handle(self, event: Doc.DocCreated, process_event): document_create = DocumentCreate( document_id=event.originator_id, body=event.body, effective_count=0 ) with get_db() as db: create_document(db, document_create)
118. class DocService(Application): is_snapshotting_enabled = True def modify(self, doc_id, body): doc = self.repository.get(doc_id) doc.modify(body) self.save(doc) self.take_snapshot(doc.id) class Doc(Aggregate): class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body class DocModified(Aggregate.Event): body: str @event(DocModified) def modify(self, body): self.body = body
119. class DocService(Application): is_snapshotting_enabled = True def modify(self, doc_id, body): doc = self.repository.get(doc_id) doc.modify(body) self.save(doc) self.take_snapshot(doc.id) class Doc(Aggregate): class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body class DocModified(Aggregate.Event): body: str @event(DocModified) def modify(self, body): self.body = body
120. class DocService(Application): is_snapshotting_enabled = True def modify(self, doc_id, body): doc = self.repository.get(doc_id) doc.modify(body) self.save(doc) self.take_snapshot(doc.id) class Doc(Aggregate): class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body class DocModified(Aggregate.Event): body: str @event(DocModified) def modify(self, body): self.body = body
121. class DocService(Application): is_snapshotting_enabled = True def modify(self, doc_id, body): doc = self.repository.get(doc_id) doc.modify(body) self.save(doc) self.take_snapshot(doc.id) class Doc(Aggregate): class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body class DocModified(Aggregate.Event): body: str @event(DocModified) def modify(self, body): self.body = body
122. class DocService(Application): is_snapshotting_enabled = True def modify(self, doc_id, body): doc = self.repository.get(doc_id) doc.modify(body) self.save(doc) self.take_snapshot(doc.id) class Doc(Aggregate): class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body class DocModified(Aggregate.Event): body: str @event(DocModified) def modify(self, body): self.body = body
123. topic: state: app.module.document.domain.models.doc:Doc.DocModified { "timestamp": { "_type_": "datetime_iso", "_data_": "2024-09-26T08:47:41.955242+00:00" }, "body": "hello world" }
131. class DocService(Application): is_snapshotting_enabled = True def modify(self, doc_id, body): doc = self.repository.get(doc_id) doc.modify(body) self.save(doc) self.take_snapshot(doc.id) class Doc(Aggregate): class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body class DocModified(Aggregate.Event): body: str @event(DocModified) def modify(self, body): self.body = body
132. class DocService(Application): is_snapshotting_enabled = True def modify(self, doc_id, body): doc = self.repository.get(doc_id) doc.modify(body) self.save(doc) self.take_snapshot(doc.id) class Doc(Aggregate): class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body class DocModified(Aggregate.Event): body: str @event(DocModified) def modify(self, body): self.body = body
133. class DocService(Application): is_snapshotting_enabled = True def modify(self, doc_id, body): doc = self.repository.get(doc_id) doc.modify(body) self.save(doc) self.take_snapshot(doc.id) class Doc(Aggregate): class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body class DocModified(Aggregate.Event): body: str @event(DocModified) def modify(self, body): self.body = body
136. class DocService(Application): is_snapshotting_enabled = True def modify(self, doc_id, body): doc = self.repository.get(doc_id) doc.modify(body) self.save(doc) self.take_snapshot(doc.id) class Doc(Aggregate): class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body class DocModified(Aggregate.Event): body: str @event(DocModified) def modify(self, body): self.body = body
137. class DocService(Application): is_snapshotting_enabled = True def modify(self, doc_id, body): doc = self.repository.get(doc_id) doc.modify(body) self.save(doc) self.take_snapshot(doc.id) class Doc(Aggregate): class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body class DocModified(Aggregate.Event): body: str @event(DocModified) def modify(self, body): self.body = body
138. class DocService(Application): is_snapshotting_enabled = True def modify(self, doc_id, body): doc = self.repository.get(doc_id) doc.modify(body) self.save(doc) self.take_snapshot(doc.id) class Doc(Aggregate): class DocCreated(Aggregate.Created): body: str @event(DocCreated) def __init__(self, body): self.body = body class DocModified(Aggregate.Event): body: str @event(DocModified) def modify(self, body): self.body = body
153. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool # Event Classes class Created(Aggregate.Created): def apply(self, aggregate: Aggregate): aggregate.when(self) @classmethod def create(cls, doc_id: UUID, user_id: UUID): return cls._create( event_class=cls.Created, id=cls.create_id(doc_id, user_id), doc_id=doc_id, user_id=user_id, ) @staticmethod def create_id(doc_id: UUID, user_id: UUID) -> UUID: return uuid5(NAMESPACE_URL, f"/effective/{doc_id}|{user_id}") @event(EffectiveCreated) def __init__(self, doc_id: UUID, user_id: UUID): self.doc_id = doc_id self.user_id = user_id self.marked = True # <略>
154. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool # Event Classes class Created(Aggregate.Created): def apply(self, aggregate: Aggregate): aggregate.when(self) @classmethod def create(cls, doc_id: UUID, user_id: UUID): return cls._create( event_class=cls.Created, id=cls.create_id(doc_id, user_id), doc_id=doc_id, user_id=user_id, ) @staticmethod def create_id(doc_id: UUID, user_id: UUID) -> UUID: return uuid5(NAMESPACE_URL, f"/effective/{doc_id}|{user_id}") @event(EffectiveCreated) def __init__(self, doc_id: UUID, user_id: UUID): self.doc_id = doc_id self.user_id = user_id self.marked = True # <略>
155. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool # Event Classes class Created(Aggregate.Created): def apply(self, aggregate: Aggregate): aggregate.when(self) @classmethod def create(cls, doc_id: UUID, user_id: UUID): return cls._create( event_class=cls.Created, id=cls.create_id(doc_id, user_id), doc_id=doc_id, user_id=user_id, ) @staticmethod def create_id(doc_id: UUID, user_id: UUID) -> UUID: return uuid5(NAMESPACE_URL, f"/effective/{doc_id}|{user_id}") @event(EffectiveCreated) def __init__(self, doc_id: UUID, user_id: UUID): self.doc_id = doc_id self.user_id = user_id self.marked = True # <略>
156. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool # Event Classes class Created(Aggregate.Created): def apply(self, aggregate: Aggregate): aggregate.when(self) @classmethod def create(cls, doc_id: UUID, user_id: UUID): return cls._create( event_class=cls.Created, id=cls.create_id(doc_id, user_id), doc_id=doc_id, user_id=user_id, ) @staticmethod def create_id(doc_id: UUID, user_id: UUID) -> UUID: return uuid5(NAMESPACE_URL, f"/effective/{doc_id}|{user_id}") @event(EffectiveCreated) def __init__(self, doc_id: UUID, user_id: UUID): self.doc_id = doc_id self.user_id = user_id self.marked = True # <略>
157. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool # Event Classes class Created(Aggregate.Created): def apply(self, aggregate: Aggregate): aggregate.when(self) @classmethod def create(cls, doc_id: UUID, user_id: UUID): return cls._create( event_class=cls.Created, id=cls.create_id(doc_id, user_id), doc_id=doc_id, user_id=user_id, ) @staticmethod def create_id(doc_id: UUID, user_id: UUID) -> UUID: return uuid5(NAMESPACE_URL, f"/effective/{doc_id}|{user_id}") @event(EffectiveCreated) def __init__(self, doc_id: UUID, user_id: UUID): self.doc_id = doc_id self.user_id = user_id self.marked = True # <略>
159. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool def mark(self): if not self.marked: self._mark(self.doc_id) @event(EffectiveMarked) def _mark(self, doc_id: UUID): self.marked = True # <略>
160. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool def mark(self): if not self.marked: self._mark(self.doc_id) @event(EffectiveMarked) def _mark(self, doc_id: UUID): self.marked = True # <略>
161. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool def mark(self): if not self.marked: self._mark(self.doc_id) @event(EffectiveMarked) def _mark(self, doc_id: UUID): self.marked = True # <略>
162. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool def mark(self): if not self.marked: self._mark(self.doc_id) @event(EffectiveMarked) def _mark(self, doc_id: UUID): self.marked = True # <略>
163. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool def mark(self): if not self.marked: self._mark(self.doc_id) @event(EffectiveMarked) def _mark(self, doc_id: UUID): self.marked = True # <略>
164. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool def mark(self): if not self.marked: self._mark(self.doc_id) @event(EffectiveMarked) def _mark(self, doc_id: UUID): self.marked = True # <略>
166. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool def unmark(self): if self.marked: self._unmark() def _unmark(self): self.trigger_event(Effective.EffectiveUnmarked, doc_id=self.doc_id) @singledispatchmethod def when(self, event) -> None: raise NotImplementedError(f"Unsupported event type: {type(event)}") @when.register(EffectiveUnmarked) def _(self, event: EffectiveUnmarked) -> None: self.marked = False # <略>
167. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool def unmark(self): if self.marked: self._unmark() def _unmark(self): self.trigger_event(Effective.EffectiveUnmarked, doc_id=self.doc_id) @singledispatchmethod def when(self, event) -> None: raise NotImplementedError(f"Unsupported event type: {type(event)}") @when.register(EffectiveUnmarked) def _(self, event: EffectiveUnmarked) -> None: self.marked = False # <略>
168. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool def unmark(self): if self.marked: self._unmark() def _unmark(self): self.trigger_event(Effective.EffectiveUnmarked, doc_id=self.doc_id) @singledispatchmethod def when(self, event) -> None: raise NotImplementedError(f"Unsupported event type: {type(event)}") @when.register(EffectiveUnmarked) def _(self, event: EffectiveUnmarked) -> None: self.marked = False # <略>
169. class Effective(Aggregate): doc_id: UUID user_id: UUID marked: bool def unmark(self): if self.marked: self._unmark() def _unmark(self): self.trigger_event(Effective.EffectiveUnmarked, doc_id=self.doc_id) @singledispatchmethod def when(self, event) -> None: raise NotImplementedError(f"Unsupported event type: {type(event)}") @when.register(EffectiveUnmarked) def _(self, event: EffectiveUnmarked) -> None: self.marked = False # <略>
171. class DocumentDataModelProjection(ProcessApplication): @singledispatchmethod def policy(self, domain_event, process_event): """Default policy""" @policy.register(Effective.EffectiveCreated) def handle(self, event: Effective.EffectiveCreated, _): with get_db() as db: document = get_document_by_id(db, event.doc_id) update_document(db, event.doc_id, DocumentUpdate(effective_count=document.effective_count + 1)) @policy.register(Effective.EffectiveMarked) def handle(self, event: Effective.EffectiveMarked, _): with get_db() as db: document = get_document_by_id(db, event.doc_id) update_document(db, event.doc_id, DocumentUpdate(effective_count=document.effective_count + 1)) @policy.register(Effective.EffectiveUnmarked) def handle(self, event: Effective.EffectiveUnmarked, _): with get_db() as db: document = get_document_by_id(db, event.doc_id) update_document(db, event.doc_id, DocumentUpdate(effective_count=document.effective_count - 1))
172. class DocumentDataModelProjection(ProcessApplication): @singledispatchmethod def policy(self, domain_event, process_event): """Default policy""" @policy.register(Effective.EffectiveCreated) def handle(self, event: Effective.EffectiveCreated, _): with get_db() as db: document = get_document_by_id(db, event.doc_id) update_document(db, event.doc_id, DocumentUpdate(effective_count=document.effective_count + 1)) @policy.register(Effective.EffectiveMarked) def handle(self, event: Effective.EffectiveMarked, _): with get_db() as db: document = get_document_by_id(db, event.doc_id) update_document(db, event.doc_id, DocumentUpdate(effective_count=document.effective_count + 1)) @policy.register(Effective.EffectiveUnmarked) def handle(self, event: Effective.EffectiveUnmarked, _): with get_db() as db: document = get_document_by_id(db, event.doc_id) update_document(db, event.doc_id, DocumentUpdate(effective_count=document.effective_count - 1))
179. os.environ['PERSISTENCE_MODULE'] = 'eventsourcing_axonserver' os.environ['AXONSERVER_URI'] = '127.0.0.1:8124' system = System( pipes=[ [DocService, DocumentDataModelProjection], [EffectiveService, DocumentDataModelProjection], ] ) runner = MultiThreadedRunner runner = runner(system, None) runner.start() doc_service = runner.get(DocService)