Coverage for packages/kwai-core/src/kwai_core/domain/presenter.py: 100%

1"""Module for defining a presenter.""" 

2 

3from abc import ABC, abstractmethod 

4from dataclasses import dataclass 

5from typing import AsyncIterator, Self 

6 

7 

8@dataclass(frozen=True, kw_only=True, slots=True) 

9class IterableResult[T]: 

10 """A dataclass used to represent a result with multiple entities.""" 

11 

12 count: int 

13 offset: int = 0 

14 limit: int = 0 

15 iterator: AsyncIterator[T] 

16 

17 def __aiter__(self): 

18 """Iterate over the entities.""" 

19 return self.iterator 

20 

21 

22class Presenter[T](ABC): 

23 """An interface for a presenter. 

24 

25 A presenter is used to transform an entity into another object that can be used 

26 in a view. 

27 

28 An example: convert to a JSON:API resource for returning the entity in a restful 

29 API. 

30 """ 

31 

32 @abstractmethod 

33 def present(self, use_case_result: T) -> Self: 

34 """Present the entity. 

35 

36 This method is responsible for converting the entity. 

37 """ 

38 

39 

40class EntityPresenter[T](Presenter[T]): 

41 """A simple presenter that stores the entity it receives.""" 

42 

43 def __init__(self): 

44 super().__init__() 

45 self._entity: T | None = None 

46 

47 @property 

48 def entity(self) -> T | None: 

49 """Return the entity.""" 

50 return self._entity 

51 

52 def present(self, entity: T) -> Self: 

53 """It just stores the received entity.""" 

54 self._entity = entity 

55 return self 

56 

57 

58class AsyncPresenter[T](ABC): 

59 """An interface for an async presenter. 

60 

61 A presenter is used to transform an entity into another object that can be used 

62 in a view. 

63 

64 An example: convert to a JSON:API resource for returning the entity in a restful 

65 API. 

66 """ 

67 

68 @abstractmethod 

69 async def present(self, use_case_result: T) -> Self: 

70 """Present the entity. 

71 

72 This method is responsible for converting the entity. 

73 """ 

74 

75 

76class CountIterableAsyncPresenter[T](AsyncPresenter[IterableResult[T]]): 

77 """A presenter that counts the number of entries in the use case result. 

78 

79 This presenter can be used in tests. 

80 """ 

81 

82 def __init__(self): 

83 super().__init__() 

84 self._count = 0 

85 

86 @property 

87 def count(self) -> int: 

88 """Return count.""" 

89 return self._count 

90 

91 async def present(self, use_case_result: IterableResult[T]) -> Self: 

92 """Process the result of the use case.""" 

93 async for _ in use_case_result: 

94 self._count += 1 

95 return self