Coverage for backend/common/src/common/models/notifications/payloads.py: 100%

1from enum import Enum 

2from typing import Literal, Optional 

3from pydantic import BaseModel, Field 

4from common.models import CloudEventsPayload 

5from ahuora_builder_types.payloads import BuildStateRequestContext 

6from ahuora_builder_types.unit_model_schema import SolvedPropertyValueSchema 

7 

8 

9class NotificationServiceMessageType(str, Enum): 

10 TASK_COMPLETED = "task/completed" 

11 TASK_UPDATED = "task/updated" 

12 TASK_CANCELLED = "task/cancelled" 

13 TASK_CANCELLING = "task/cancelling" 

14 TASK_CHILDREN_CANCELLED = "task/children-cancelled" 

15 BUILD_STATE_COMPLETED = "build-state/completed" 

16 RESULT_SUMMARY_CALCULATING = "result-summary/calculating" 

17 RESULT_SUMMARY_AVAILABLE = "result-summary/available" 

18 RESULT_SUMMARY_FAILED = "result-summary/failed" 

19 AUTO_REVISION_SKIPPED_ACTIVE_TASKS = ( 

20 "flowsheet-version/auto-save-skipped-active-tasks" 

21 ) 

22 

23class NotificationServiceMessage(BaseModel): 

24 message_type: NotificationServiceMessageType 

25 data: dict 

26 

27# For use in telling the notification (Django Channels) service to send a notification 

28class NotificationServiceEnvelope(BaseModel): 

29 flowsheet_id: int 

30 messages: list[NotificationServiceMessage] 

31 

32NotificationEvent = CloudEventsPayload[NotificationServiceEnvelope] 

33 

34# The payload the user is sent by the notification service 

35class TaskCompletedPayload(BaseModel): 

36 task_id: int 

37 

38 

39class BuildStateCompletedPayload(BaseModel): 

40 context: BuildStateRequestContext 

41 status: Literal["success", "error"] 

42 properties: Optional[list[SolvedPropertyValueSchema]] = None 

43 error: Optional[str] = None 

44 failure_kind: Optional[Literal["idaes-error", "delivery-failure"]] = None 

45 

46 

47class ResultSummaryCalculationError(BaseModel): 

48 message: str 

49 cause: str 

50 simulation_object_id: Optional[int] = None 

51 result_key: Optional[str] = None 

52 property_value_id: Optional[int] = None 

53 

54 

55class ResultSummaryCalculationStatus(str, Enum): 

56 pending = "pending" 

57 calculating = "calculating" 

58 available = "available" 

59 partial = "partial" 

60 failed = "failed" 

61 

62 

63class ResultSummaryStatusPayload(BaseModel): 

64 scenario_id: int 

65 simulation_object_id: int 

66 source_task_id: Optional[int] 

67 status: ResultSummaryCalculationStatus 

68 errors: list[ResultSummaryCalculationError] = Field(default_factory=list) 

69 

70 

71class ResultSummaryCalculatingPayload(BaseModel): 

72 scenario_id: int 

73 simulation_object_ids: list[int] 

74 source_task_id: int 

75 

76 

77class ResultSummaryAvailablePayload(BaseModel): 

78 scenario_id: int 

79 simulation_object_id: int 

80 source_task_id: int 

81 errors: list[ResultSummaryCalculationError] = Field(default_factory=list) 

82 

83 

84class ResultSummaryFailedPayload(BaseModel): 

85 scenario_id: int 

86 simulation_object_id: int 

87 source_task_id: int 

88 errors: list[ResultSummaryCalculationError]