Coverage for rfpy/model/notes.py: 100%

1from enum import Enum 

2from datetime import datetime 

3from typing import Optional, TYPE_CHECKING 

4 

5from sqlalchemy import text, Integer, ForeignKey, DateTime, Boolean, Index 

6from sqlalchemy.orm import relationship, foreign, remote, Mapped, mapped_column 

7from sqlalchemy.dialects.mysql import VARCHAR, LONGTEXT, TINYINT 

8import sqlalchemy.types as types 

9 

10 

11from rfpy.model.meta import Base 

12from rfpy.model import Organisation 

13 

14if TYPE_CHECKING: 

15 from rfpy.model.project import Project 

16 

17kinds = {0: "IssuerNote", 1: "RespondentNote"} 

18kinds_int = {v: k for k, v in kinds.items()} 

19 

20 

21class NoteKind(types.TypeDecorator): 

22 impl = TINYINT(2) 

23 

24 cache_ok = True 

25 

26 def process_bind_param(self, value, dialect): 

27 return kinds_int[value] 

28 

29 def process_result_value(self, value, dialect): 

30 return kinds[value] 

31 

32 

33class Distribution(str, Enum): 

34 BROADCAST_NOTICE = "BROADCAST_NOTICE" 

35 RESPONDENT_QUERY = "RESPONDENT_QUERY" 

36 RESPONDENT_INTERNAL_MEMO = "RESPONDENT_INTERNAL_MEMO" 

37 ISSUER_INTERNAL_MEMO = "ISSUER_INTERNAL_MEMO" 

38 TARGETED = "TARGETED" 

39 

40 

41class ProjectNote(Base): 

42 __tablename__ = "project_notes" 

43 __table_args__ = ( 

44 Index("ft_notes", "note_text", mysql_prefix="FULLTEXT"), 

45 ) + Base.__table_args__ 

46 

47 public_attrs = ( 

48 "id,note_time,user_id,org_id,target_org_id,note_text,private,kind" 

49 ).split(",") 

50 project_id: Mapped[int] = mapped_column( 

51 Integer, 

52 ForeignKey("projects.id", name="project_notes_ibfk_3", ondelete="CASCADE"), 

53 nullable=False, 

54 ) 

55 

56 note_time: Mapped[datetime] = mapped_column( 

57 DateTime, 

58 nullable=False, 

59 server_default=text("CURRENT_TIMESTAMP"), 

60 default=datetime.now, 

61 ) 

62 user_id: Mapped[str] = mapped_column( 

63 VARCHAR(length=150), nullable=False, server_default=text("''") 

64 ) 

65 org_id: Mapped[Optional[str]] = mapped_column( 

66 VARCHAR(length=150), 

67 ForeignKey("organisations.id", ondelete="SET NULL", onupdate="CASCADE"), 

68 nullable=True, 

69 server_default=text("''"), 

70 ) 

71 target_org_id: Mapped[Optional[str]] = mapped_column( 

72 VARCHAR(length=150), 

73 ForeignKey("organisations.id", ondelete="SET NULL", onupdate="CASCADE"), 

74 nullable=True, 

75 ) 

76 note_text: Mapped[str] = mapped_column(LONGTEXT(), nullable=False) 

77 private: Mapped[bool] = mapped_column( 

78 Boolean, nullable=False, server_default=text("'0'") 

79 ) 

80 kind: Mapped[str] = mapped_column( 

81 "type", NoteKind, nullable=False, server_default=text("'0'") 

82 ) 

83 

84 organisation: Mapped[Optional[Organisation]] = relationship( 

85 "Organisation", 

86 primaryjoin=foreign(org_id) == remote(Organisation.id), 

87 uselist=False, 

88 ) 

89 

90 target_organisation: Mapped[Optional[Organisation]] = relationship( 

91 "Organisation", 

92 primaryjoin=foreign(target_org_id) == remote(Organisation.id), 

93 uselist=False, 

94 ) 

95 

96 project: Mapped["Project"] = relationship( 

97 "Project", 

98 back_populates="notes_query", 

99 ) 

100 

101 def __repr__(self): 

102 tm = "<Note[%s] kind: %s, private: %s, target_org_id: %s>" 

103 ms = tm % (self.id, self.kind, self.private, self.target_org_id) 

104 return ms 

105 

106 @property 

107 def distribution(self) -> Distribution: 

108 if self.private: 

109 if self.kind == "IssuerNote": 

110 return Distribution.ISSUER_INTERNAL_MEMO 

111 elif self.kind == "RespondentNote": 

112 return Distribution.RESPONDENT_INTERNAL_MEMO 

113 elif self.kind == "IssuerNote": 

114 if self.target_org_id is None: 

115 return Distribution.BROADCAST_NOTICE 

116 else: 

117 return Distribution.TARGETED 

118 elif self.kind == "RespondentNote": 

119 return ( 

120 Distribution.RESPONDENT_QUERY 

121 ) # Note isn't private: can be viewed by buyer 

122 

123 raise ValueError("ProjectNote misconfigured %s", self) 

124 

125 @property 

126 def pretty_distribution(self) -> str: 

127 return " ".join(a.title() for a in self.distribution.value.split("_"))