Trải nghiệm mô hình Clean Architecture với Ứng dụng Todos
Một ví dụ minh họa việc áp dụng mô hình Clean Architecture trong Python3
Bài viết này là phần thứ ba nằm trong blog series khai phá Domain-Driven Design. Bạn có thể bấm vào đây để xem danh sách toàn bộ các bài viết.
Ở phần 2, khi đề cập về lý thuyết KTPT (kiến trúc phân tầng/layer architecture) trong Domain-driven Design, mình có lấy ví dụ về ứng dụng Todos và dừng lại ở việc nêu ra vài yêu cầu nghiệp vụ giả định. Tại bài viết này, mình sẽ chia sẻ chi tiết về quá trình design — code — refactor ứng dụng này để demo việc áp dụng lý thuyết KTPT vào project cụ thể để xem nó sẽ diễn ra như thế nào.
Nhắc lại về yêu cầu nghiệp vụ của bài toán Todos:
- User chỉ cần nhập username để login (ko cần mật khẩu, ko cần đăng ký);
- User có thể quản lý một hoặc nhiều task;
- Trạng thái mỗi task có thể là: new ➙ doing ➙ done;
- Mỗi user tại một thời điểm chỉ được phép có tối đa 3 task với trạng thái là doing. Nếu tính từ lúc bắt đầu sử dụng phần mềm, người này đã từng hoàn thành ít nhất 3 task thì mới được quyền có tối đa 5 task đồng thời có trạng thái là doing.
Với y/c nghiệp vụ như trên, SA (software architecture) đưa ra bản thiết kế domain model như sau:
Thời điểm cả team chốt được bản thiết kế domain model cũng chính là lúc a/e developer có thể xắn tay áo lên và code được rồi. Thứ tự coding sẽ theo hướng: lập trình layer bên trong trước, sau đó lập trình lan ra các layer bên ngoài. Để dễ kiểm soát lỗi, mình khuyên ko nên lập trình full layer này rồi mới chuyển sang layer khác. Thay vào đó, hãy chọn một vài UCs (use cases) mà bạn muốn xử lý, sau đó mới lập trình các layer từ trong ra ngoài sao cho vừa đủ để đáp ứng các UCs này.
TL;DR
Mã nguồn demo của ứng dụng được host trên GitHub, các bạn có thể clone về để xem chi tiết.
Bonus thêm cho a/e Java & Golang:
- https://github.com/hoangtuan151/todos-ddd-java-example
- https://github.com/hoangtuan151/todos-ddd-golang-example
Round 1
Hãy bắt đầu với 2 UCs đơn giản của Todos application:
- User tạo task mới;
- User lấy danh sách các task mà mình đã tạo.
Domain layer
class Task:
def __init__(self, tid, desc, state=None):
self._tid = tid
self._desc = desc
self._state = state if state is not None else 'NEW' def __str__(self):
return 'Task(tid: {0}, state: {1}, desc: {2})'.format(
self._tid,
self._state,
self._desc
)
Trong lớp Task, phương thức __init__() giúp tạo instance object từ 3 tham số: task id, task description và state (tùy chọn). Phương thức __str__() in ra nội dung của task object, giúp việc debug thuận tiện hơn.
UCs layer
class TaskUC:
def __init__(self, task_repo):
self._task_repo = task_repo def create_task(self, username, desc):
tid = str(uuid.uuid4())
new_task = Task(tid, desc)
self._task_repo.create_task_for_user(username, new_task)
return new_task def get_task_list(self, username):
return self._task_repo.get_tasks_by_username(username)
Trong bài toán Todos giả định này, chúng ta chỉ có các nghiệp vụ liên quan đến task và chúng đều được implement trong TaskUC class.
Theo lý thuyết KTPT, tầng UCs sẽ ko kết nối trực tiếp với các hệ thống Database cụ thể mà sẽ kết nối thông qua interface. Ở đây bạn có thể thấy TaskUC sử dụng đối tượng dependency là task_repo (task repository) như một interface để kết nối đến Database cụ thể sau này. Lưu ý là trong Python ko có khái niệm interface, nên task_repo có thể là bất kỳ instance object nào, miễn là nó có các phương thức cần thiết (create_task_for_user() và get_tasks_by_username()).
Hai nghiệp vụ được implement ở đây khá simple:
create_task()sử dụng lớpTasktừ domain layer để tạo và truyền object instance của lớp này cho hàm_task_repo.create_task_for_user()để xử lý tạo task mới;get_task_list()thậm chí còn đơn giản hơn: chỉ sử dụng_task_repo.get_tasks_by_username()để lấy danh sách task cho user.
External layer — Database
Tầng này sẽ cài đặt interface được định nghĩa/sử dụng ở tầng trước đó (chính là task_repo). Để tránh sa đà vào việc setup/kết nối đến một hệ thống DB phức tạp gây mất thời gian, tạm thời chúng ta sẽ sử dụng bộ nhớ RAM làm DB. Sau này khi code xong tầng domain & UCs, chúng ta sẽ thay thế RAM DB bởi SQLite DB.
class MemRepos:
def __init__(self):
self._memory = {} def create_task_for_user(self, username, new_task):
if username not in self._memory:
self._memory[username] = []
self._memory[username].append(new_task) def get_tasks_by_username(self, username):
if username not in self._memory:
return []
else:
return self._memory[username]
Class MemRepos sử dụng Python dict để lưu trữ dữ liệu theo cấu trúc:
{
'user1': [Object<task1>, Object<task2>],
'user2': [Object<task1>]
}Theo cấu trúc này thì việc implement 2 hàm create_task_for_user() và get_tasks_by_username() như đoạn code trên khá là straight forward rồi!
External layer — Application
Cho đến lúc này, chúng ta đã code xong các cấu phần của 2 nghiệp vụ (tạo task mới & lấy danh sách các task) cho cả 3 layer. Tiếp theo sẽ đến tầng ứng dụng — là nơi tiếp nhận user input, rồi sử dụng các đối tượng UCs để thực thi nghiệp vụ và show kết quả cho user. Để tiết kiệm thời gian, chúng ta sẽ hướng đến việc tạo ra chương trình command line (CLI) đơn giản đủ để thể hiện 2 nghiệp vụ nói trên.
Các bạn có thể xem mã nguồn đầy đủ trên GitHub, ở đây mình chỉ nêu ra một số dòng code để demo việc sử dụng các đối tượng thuộc layer bên trong tại application layer:
# Import các đối tượng từ inner layer
from todos.domain.task import Task
from todos.repos.mem_repos import MemRepos
from todos.usecases.task_uc import TaskUC# Khởi tạo repository
REPO = MemRepos()#
# Tạo task
#
def add_task():
# Lấy user input
desc = sys.stdin.readline()
desc = desc.strip() # Sử dụng UC object để thực thi nghiệp vụ
task_uc = TaskUC(REPO)
result = task_uc.create_task(USER, desc) # Print result...
#
# Lấy danh sách task
#
def list_tasks():
# Sử dụng UC object để thực thi nghiệp vụ
task_uc = TaskUC(REPO)
task_list = task_uc.get_task_list(USER) # Print result...
Demo
Closing Thoughts
Qua việc implement 2 UCs đầu tiên cho ứng dụng Todos, hi vọng các bạn có feeling thực tế về việc coding các layer trong KTPT và thấy được:
- Việc phân chia layer một cách clean & theo rule giúp cho mã nguồn của các layer có sự độc lập nhất định, qua đó việc apply unit test vào từng layer sẽ dễ dàng hơn;
- Code của tầng ứng dụng sáng sủa hơn vì chỉ chứa tỉ lệ nhỏ mã nguồn sử dụng các đối tượng của các layer bên trong và ko còn bị lẫn lộn hay lỗn lận bởi các đoạn code truy cập database hoặc code của tầng nghiệp vụ. Do vậy nếu sau này bạn có sửa nghiệp vụ hoặc muốn thay mới tầng giao diện thì cũng sẽ rất dễ dàng để xử lý.
Round 2
Với mục tiêu giúp các bạn làm quen với KTPT, round 1 chọn implement 2 nghiệp vụ rất đơn giản nên mã nguồn cho layer domain chưa nhiều. Sang round 2, chúng ta sẽ implement nốt các nghiệp vụ còn lại và dành nhiều sự quan tâm hơn cho domain layer, qua đó các bạn sẽ có cái nhìn sâu hơn về domain — một nội dung trọng tâm trong Domain-driven Design.
Trong 2 nghiệp vụ còn lại:
- Đánh dấu task có trạng thái done;
- Đánh dấu task có trạng thái doing,
mình sẽ chỉ trình bày về nghiệp vụ thứ 2 vì nó chứa đựng business logic cần được mô hình hóa từ real life vào software. Nghiệp vụ đầu tiên ko khó, các bạn xem qua code là có thể hiểu được.
Domain layer
Lớp Task được bổ sung vài phương thức/thuộc tính tiện ích giúp đóng gói một số khái niệm:
class Task:
...
@property
def tid(self):
return self._tid
...
def is_done(self):
return self._state == 'DONE' def is_doing(self):
return self._state == 'DOING' def mark_doing(self):
self._state = 'DOING' def mark_done(self):
self._state = 'DONE'
Các bạn có thể thấy, việc sử dụng chuỗi string DONE hay DOING để đại diện cho trạng thái của task được encapsulate hoàn toàn trong lớp Task, mã nguồn sử dụng lớp Task chỉ việc gọi các phương thức is_done() hay is_doing() để kiểm tra là đủ. Giả sử sau này vì lý do nào đó, trạng thái của task được đại diện bởi number thay vì string (chẳng hạn: 1 là done, 2 là doing) thì chỉ cần sửa lại cách cài đặt phương thức của lớp Task là đủ, các đoạn code đang có bên ngoài ko cần phải thay đổi theo.
Lớp User được thêm mới:
class User:
def __init__(self, username):
self._username = username
self._total_done_task = 0
self._total_doing_task = 0 ... def __str__(self):
return 'Task(User: {0}, done: {1}, doing: {2})'.format(
self._username,
self._total_done_task,
self._total_doing_task
)
Lớp MaxConcurrencyDoingTaskPolicy được thêm mới:
class MaxConcurrencyDoingTaskPolicy:
def __init__(self, user):
self._threshold = 3
self._current_doing_task = user.doing_task
if user.done_task >= 3:
self._threshold = 5 def allow_next_doing_task(self):
return self._current_doing_task < self._threshold
Đây chính là nơi đóng gói logic mà bạn đã thấy ở mục user requirement: “mỗi user tại một thời điểm chỉ được phép có tối đa 3 task với trạng thái là doing. Nếu tính từ lúc bắt đầu sử dụng phần mềm, người này đã từng hoàn thành ít nhất 3 task thì mới được quyền có tối đa 5 task đồng thời có trạng thái là doing.”
UCs layer
Lớp TaskUC được sửa đổi, bổ sung để đáp ứng nghiệp vụ mới:
class TaskUC:
def __init__(self, task_repo, user_repo=None):
self._task_repo = task_repo
self._user_repo = user_repo def mark_task_as_doing(self, username, task_id):
user = self._user_repo.find_user_by_username(username)
management_policy = MaxConcurrencyDoingTaskPolicy(user)
if management_policy.allow_next_doing_task():
self._task_repo.mark_task_as_doing(username, task_id)
else:
raise Exception('Over threshold')
...
Lớp MaxConcurrencyDoingTaskPolicy được sử dụng ở đây để control xem liệu user có được phép đánh dấu một task là doing hay ko (pattern này được gọi là STRATEGY). Sau này nếu có thay đổi nghiệp vụ chút ít thì chỉ cần sửa lại logic của hàm allow_next_doing_task() là đủ.
Nếu là trước đây, có thể mình sẽ code nghiệp vụ trên như sau (dạng pseudocode):
def pseudo_mark_task_as_doing(self, username, task_id):
user = database.find_user_by_username(username)
int threshold = 3
if user.done_task >= 3:
threshold = 5
if user.doing_task < threshold:
database.mark_task_as_doing(username, task_id)
else:
raise Exception('Over threshold')Nhìn qua đã thấy sự khác biệt fải ko nào, đoạn code này rối rắm hơn chút xíu do đã nhúng thêm logic nghiệp vụ. Tất nhiên vì nghiệp vụ này cũng ko quá phức tạp, nên bạn vẫn cảm thấy có thể manage được đoạn code trên. Tuy vậy trong thực tế, khi gặp những case có nghiệp vụ phức tạp và trải qua 2–3 lần thay đổi business requirement thì sớm thôi bạn sẽ thay đổi quan điểm của mình.
External layer — Database
Lớp MemRepos được bổ sung:
class MemRepos:
def find_user_by_username(self, username):
tasks = self.get_tasks_by_username(username)
done_count = 0
doing_count = 0
for task in tasks:
if task.is_done():
done_count += 1
elif task.is_doing():
doing_count += 1 user = User(username)
user.doing_task = doing_count
user.done_task = done_count
return user def mark_task_as_doing(self, username, task_id):
tasks = self.get_tasks_by_username(username)
for task in tasks:
if task.tid == task_id:
task.mark_doing()
break
Demo
Round 3
Qua 2 round đầu, bạn đã có thể nắm bắt phần nào về việc coding các layer trong KTPT và hiểu hơn về domain trong DDD. Ở round 3 này, chúng ta sẽ thay thế RAM DB bằng SQLite DB để thấy được khi áp dụng đúng KTPT, việc apply change sẽ trở nên dễ dàng hơn nhiều.
Đây là lược đồ quan hệ (DB schema) mà chúng ta sẽ thiết kế cho SQLite DB:
Domain layer
Lớp Task được bổ sung vài thuộc tính tiện ích:
class Task:
...
@property
def tid(self):
return self._tid
@property
def desc(self):
return self._desc @property
def state(self):
return self._state
...
External layer — Database
Bổ sung lớp SQLiteRepos với các phương thức chính (có signature giống như các phương thức của class MemRepos):
- Tạo task mới cho user:
def create_task_for_user(self, username, new_task):
sql = '''
INSERT INTO
Task (id, username, desc, state)
VALUES(?, ?, ?, ?)
'''
param = (new_task.tid, username, new_task.desc, new_task.state)
cur = self._db_conn.cursor()
cur.execute(sql, param)
self._db_conn.commit()
print('debug sql:', cur.rowcount)2. Lấy danh sách task:
def get_tasks_by_username(self, username):
sql = '''SELECT * FROM Task WHERE username = ?'''
cur = self._db_conn.cursor()
cur.execute(sql, (username,))
return cur.fetchall()3. Tìm user:
def find_user_by_username(self, username):
sql = '''
SELECT state, count(id) as num
FROM Task
WHERE username = ?
GROUP BY state
'''
cur = self._db_conn.cursor()
cur.execute(sql, (username,))
user = User(username)
for row in cur.fetchall():
if row[0] == 'DONE':
user.done_task = row[1]
elif row[0] == 'DOING':
user.doing_task = row[1]
return user4. Đánh dấu doing/done:
def mark_task_as_doing(self, username, task_id):
self._set_task_state(username, task_id, 'DOING')def mark_task_as_done(self, username, task_id):
self._set_task_state(username, task_id, 'DONE')def _set_task_state(self, username, task_id, new_state):
sql = '''
UPDATE Task SET state = ? WHERE username = ? and id = ?
'''
cur = self._db_conn.cursor()
cur.execute(sql, (new_state, username, task_id))
self._db_conn.commit()
External layer — Application
Chương trình CLI chỉ cần thay đổi nhỏ:
from todos.repos.sqlite_repos import SQLiteRepos
# REPO = MemRepos()
REPO = SQLiteRepos('./sqlitedb/sqlite.db')Trong khi đó, như bạn đã thấy mã nguồn của tầng domain và UCs hầu như không thay đổi.
Demo
Lời Kết
Vậy là qua 3 round, mình đã minh họa quá trình coding Todos App theo KTPT trên nền bản thiết kế domain model và minh họa việc thay đổi DB (external layer) sẽ được thực hiện ntn. Do bài viết đã khá dài, các bạn vui lòng checkout GitHub để xem các quá trình còn lại:
- Bổ sung tầng Application (REST endpoint trên nền Flask);
- Mở rộng nghiệp vụ, tăng độ phức tạp và sửa code tầng domain layer để thỏa mãn sự thay đổi này;
- Thay thế SQLite bằng MongoDB để thấy rõ hơn sự linh hoạt của mô hình KTPT (các DB khác nhau có thể được design schema khác nhau sao cho tận dụng tối đa ưu điểm của DB đó).
Cám ơn các bạn đã theo dõi! Hãy follow tài khoản này để nhận thông tin về các bài viết tiếp theo nhé.
Happy DDD \(^_^)/
