Coverage for src/kwai/core/domain/use_case.py: 90%

1"""Module for defining common classes, functions, ... for use cases.""" 

2 

3from abc import ABC, abstractmethod 

4from dataclasses import dataclass 

5from typing import AsyncIterator, NamedTuple 

6 

7 

8class UseCaseResult(ABC): 

9 """Base class for a use case result.""" 

10 

11 @abstractmethod 

12 def to_message(self) -> str: 

13 """Return a message from the result.""" 

14 raise NotImplementedError 

15 

16 

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

18class NotFoundResult[T](UseCaseResult): 

19 """A result that indicates that an entity was not found.""" 

20 

21 entity_name: str 

22 key: T 

23 

24 def to_message(self) -> str: 

25 return f"{self.entity_name} with key '{self.key}' not found." 

26 

27 

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

29class EntitiesResult[T](UseCaseResult): 

30 """A result that returns an iterator for entities and the number of entities.""" 

31 

32 count: int 

33 entities: AsyncIterator[T] 

34 

35 def to_message(self) -> str: 

36 return f"{self.count} entities." 

37 

38 

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

40class EntityResult[T](UseCaseResult): 

41 """A result that returns an entity.""" 

42 

43 entity: T 

44 

45 def to_message(self) -> str: 

46 return str(self.entity) 

47 

48 

49class UseCaseBrowseResult(NamedTuple): 

50 """A named tuple for a use case that returns a result of browsing entities.""" 

51 

52 count: int 

53 iterator: AsyncIterator 

54 

55 

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

57class TextCommand: 

58 """Input for a text.""" 

59 

60 locale: str 

61 format: str 

62 title: str 

63 summary: str 

64 content: str