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

1from enum import Enum 

2from datetime import datetime 

3 

4from sqlalchemy import ( 

5 Column, text, Integer, ForeignKey, DateTime, Boolean, Index 

6) 

7from sqlalchemy.orm import relationship, backref, foreign, remote 

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

9import sqlalchemy.types as types 

10 

11 

12from rfpy.model.meta import Base 

13from rfpy.model import Organisation 

14 

15kinds = { 

16 0: 'IssuerNote', 

17 1: 'RespondentNote' 

18} 

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

20 

21 

22class NoteKind(types.TypeDecorator): 

23 

24 impl = TINYINT(2) 

25 

26 cache_ok = True 

27 

28 def process_bind_param(self, value, dialect): 

29 return kinds_int[value] 

30 

31 def process_result_value(self, value, dialect): 

32 return kinds[value] 

33 

34 

35class Distribution(str, Enum): 

36 BROADCAST_NOTICE = "BROADCAST_NOTICE" 

37 RESPONDENT_QUERY = "RESPONDENT_QUERY" 

38 RESPONDENT_INTERNAL_MEMO = "RESPONDENT_INTERNAL_MEMO" 

39 ISSUER_INTERNAL_MEMO = "ISSUER_INTERNAL_MEMO" 

40 TARGETED = "TARGETED" 

41 

42 

43class ProjectNote(Base): 

44 __tablename__ = 'project_notes' 

45 __table_args__ = ( 

46 Index('ft_notes', 'note_text', mysql_prefix='FULLTEXT'), 

47 ) + Base.__table_args__ 

48 

49 public_attrs = ('id,note_time,user_id,org_id,target_org_id,note_text,private,kind').split(',') 

50 project_id = Column(Integer, 

51 ForeignKey('projects.id', name='project_notes_ibfk_3', ondelete='CASCADE'), 

52 nullable=False) 

53 

54 note_time = Column(DateTime, nullable=False, 

55 server_default=text("CURRENT_TIMESTAMP"), 

56 default=datetime.now) 

57 user_id = Column(VARCHAR(150), nullable=False, server_default=text("''")) 

58 org_id = Column( 

59 VARCHAR(150), 

60 ForeignKey('organisations.id', ondelete='SET NULL', onupdate='CASCADE'), 

61 nullable=True, 

62 server_default=text("''") 

63 ) 

64 target_org_id = Column( 

65 VARCHAR(150), 

66 ForeignKey('organisations.id', ondelete='SET NULL', onupdate='CASCADE'), 

67 nullable=True 

68 ) 

69 note_text = Column(LONGTEXT(), nullable=False) 

70 private = Column(Boolean, nullable=False, server_default=text("'0'")) 

71 kind = Column('type', NoteKind, nullable=False, server_default=text("'0'")) 

72 

73 organisation = relationship('Organisation', 

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

75 uselist=False) 

76 

77 target_organisation = \ 

78 relationship('Organisation', 

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

80 uselist=False) 

81 

82 project = relationship('Project', 

83 backref=backref('notes_query', 

84 lazy='dynamic', 

85 cascade='all,delete', 

86 passive_deletes=True)) 

87 

88 def __repr__(self): 

89 tm = '<Note[%s] kind: %s, private: %s, target_org_id: %s>' 

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

91 return ms 

92 

93 @property 

94 def distribution(self): 

95 if self.private: 

96 if self.kind == "IssuerNote": 

97 return Distribution.ISSUER_INTERNAL_MEMO 

98 elif self.kind == 'RespondentNote': 

99 return Distribution.RESPONDENT_INTERNAL_MEMO 

100 elif self.kind == 'IssuerNote': 

101 if self.target_org_id is None: 

102 return Distribution.BROADCAST_NOTICE 

103 else: 

104 return Distribution.TARGETED 

105 elif self.kind == 'RespondentNote': 

106 return Distribution.RESPONDENT_QUERY # Note isn't private: can be viewed by buyer 

107 

108 raise ValueError('ProjectNote misconfigured %s', self) 

109 

110 @property 

111 def pretty_distribution(self): 

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