-
Notifications
You must be signed in to change notification settings - Fork 69
/
book_command_usecase.py
121 lines (90 loc) · 3.29 KB
/
book_command_usecase.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
from abc import ABC, abstractmethod
from typing import Optional, cast
import shortuuid
from dddpy.domain.book import (
Book,
BookIsbnAlreadyExistsError,
BookNotFoundError,
BookRepository,
Isbn,
)
from .book_command_model import BookCreateModel, BookUpdateModel
from .book_query_model import BookReadModel
class BookCommandUseCaseUnitOfWork(ABC):
"""BookCommandUseCaseUnitOfWork defines an interface based on Unit of Work pattern."""
book_repository: BookRepository
@abstractmethod
def begin(self):
raise NotImplementedError
@abstractmethod
def commit(self):
raise NotImplementedError
@abstractmethod
def rollback(self):
raise NotImplementedError
class BookCommandUseCase(ABC):
"""BookCommandUseCase defines a command usecase inteface related Book entity."""
@abstractmethod
def create_book(self, data: BookCreateModel) -> Optional[BookReadModel]:
raise NotImplementedError
@abstractmethod
def update_book(
self, book_id: str, data: BookUpdateModel
) -> Optional[BookReadModel]:
raise NotImplementedError
@abstractmethod
def delete_book_by_id(self, book_id: str):
raise NotImplementedError
class BookCommandUseCaseImpl(BookCommandUseCase):
"""BookCommandUseCaseImpl implements a command usecases related Book entity."""
def __init__(
self,
uow: BookCommandUseCaseUnitOfWork,
):
self.uow: BookCommandUseCaseUnitOfWork = uow
def create_book(self, data: BookCreateModel) -> Optional[BookReadModel]:
try:
uuid = shortuuid.uuid()
isbn = Isbn(data.isbn)
book = Book(book_id=uuid, isbn=isbn, title=data.title, page=data.page)
existing_book = self.uow.book_repository.find_by_isbn(isbn.value)
if existing_book is not None:
raise BookIsbnAlreadyExistsError
self.uow.book_repository.create(book)
self.uow.commit()
created_book = self.uow.book_repository.find_by_id(uuid)
except:
self.uow.rollback()
raise
return BookReadModel.from_entity(cast(Book, created_book))
def update_book(
self, book_id: str, data: BookUpdateModel
) -> Optional[BookReadModel]:
try:
existing_book = self.uow.book_repository.find_by_id(book_id)
if existing_book is None:
raise BookNotFoundError
book = Book(
book_id=book_id,
isbn=existing_book.isbn,
title=data.title,
page=data.page,
read_page=data.read_page,
)
self.uow.book_repository.update(book)
updated_book = self.uow.book_repository.find_by_id(book.book_id)
self.uow.commit()
except:
self.uow.rollback()
raise
return BookReadModel.from_entity(cast(Book, updated_book))
def delete_book_by_id(self, book_id: str):
try:
existing_book = self.uow.book_repository.find_by_id(book_id)
if existing_book is None:
raise BookNotFoundError
self.uow.book_repository.delete_by_id(book_id)
self.uow.commit()
except:
self.uow.rollback()
raise