英文:
Event Sourcing with Python: How do I make a projection?
问题
I'm trying to learn the event sourcing design pattern. And I am using the python eventsourcing library.
I've worked through the "Dog School" tutorial, but it isnt clear to me how to make a projection. I assume I need a projection if I want to retrieve a list of items in my aggregate?
It seems like projection is not in the library. Am I responsible for writing the queries to pull the data from persistence?
From the tutorial:
class DogSchool(Application):
def register_dog(self, name):
dog = Dog(name)
self.save(dog)
return dog.id
def add_trick(self, dog_id, trick):
dog = self.repository.get(dog_id)
dog.add_trick(trick=trick)
self.save(dog)
def get_dog(self, dog_id):
dog = self.repository.get(dog_id)
return {'name': dog.name, 'tricks': tuple(dog.tricks)}
And I want a function like:
def list_dogs(self):
dogs = get_all_from_aggregate() #??
return dogs
英文:
I'm trying to learn the event sourcing design pattern. And I am using the python eventsourcing library.
I've worked through the "Dog School" tutorial, but it isnt clear to me how to make a projection. I assume I need a projection if I want to retrieve a list of items in my aggregate?
It seems like projection is not in the library. Am I responsible for writing the queries to pull the data from persistence?
From the tutorial:
class DogSchool(Application):
def register_dog(self, name):
dog = Dog(name)
self.save(dog)
return dog.id
def add_trick(self, dog_id, trick):
dog = self.repository.get(dog_id)
dog.add_trick(trick=trick)
self.save(dog)
def get_dog(self, dog_id):
dog = self.repository.get(dog_id)
return {'name': dog.name, 'tricks': tuple(dog.tricks)}
And I want a function like:
def list_dogs(self):
dogs = get_all_from_aggregate() #??
return dogs
答案1
得分: 1
以下是翻译好的部分:
不需要投影来完成这个任务,但你可以考虑以这种方式来完成。
文档提供了一个内容管理示例,展示了实现list_dogs()
方法所需的基本逻辑,而不需要创建投影。
在示例中,以下方法执行了重要的工作:
def _get_page_by_id(self, page_id: UUID) -> Page:
return cast(Page, self.repository.get(page_id))
def get_pages(
self,
gt: Optional[int] = None,
lte: Optional[int] = None,
desc: bool = False,
limit: Optional[int] = None,
) -> Iterator[PageDetailsType]:
for page_logged in self.page_log.get(gt, lte, desc, limit):
page = self._get_page_by_id(page_logged.page_id)
yield self._details_from_page(page)
如果你按照示例的方式操作,应该能够创建你的list_dogs()
方法。
英文:
You don't need a projection to do this, but you could conceivably do it that way.
The documentation gives a Content management example that shows the basic the logic necessary to implement a list_dogs()
method, without making a projection.
The heavy lifting is done by these methods in the example:
def _get_page_by_id(self, page_id: UUID) -> Page:
return cast(Page, self.repository.get(page_id))
def get_pages(
self,
gt: Optional[int] = None,
lte: Optional[int] = None,
desc: bool = False,
limit: Optional[int] = None,
) -> Iterator[PageDetailsType]:
for page_logged in self.page_log.get(gt, lte, desc, limit):
page = self._get_page_by_id(page_logged.page_id)
yield self._details_from_page(page)
If you follow the example you should be able to create your list_dogs()
method.
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论