Coverage for postrfp/model/graph.py: 96%

1from typing import Optional 

2 

3from sqlalchemy import ForeignKey 

4from sqlalchemy.orm import relationship, Mapped, mapped_column 

5from sqlalchemy.types import VARCHAR 

6from sqlalchemy.sql.sqltypes import Integer 

7 

8from .humans import Organisation 

9from .meta import Base 

10 

11 

12class RelationshipType(Base): 

13 __tablename__ = "relationship_types" 

14 

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

16 VARCHAR(length=50), 

17 ForeignKey("organisations.id", onupdate="CASCADE", ondelete="CASCADE"), 

18 nullable=True, 

19 ) 

20 name: Mapped[str] = mapped_column(VARCHAR(length=128)) 

21 description: Mapped[Optional[str]] = mapped_column( 

22 VARCHAR(length=256), nullable=True 

23 ) 

24 

25 organisation = relationship( 

26 Organisation, 

27 primaryjoin=org_id == Organisation.id, 

28 back_populates="relationship_types", 

29 ) 

30 

31 edges = relationship( 

32 "Edge", 

33 back_populates="relationship_type", 

34 cascade="all, delete", 

35 passive_deletes=True, 

36 ) 

37 

38 def __repr__(self) -> str: 

39 return f'<RelationshipType - "{self.name}" #{self.id}>' 

40 

41 

42class Edge(Base): 

43 __tablename__ = "edges" 

44 

45 id = None # type: ignore 

46 

47 from_org_id: Mapped[str] = mapped_column( 

48 VARCHAR(length=50), 

49 ForeignKey("organisations.id", onupdate="CASCADE"), 

50 primary_key=True, 

51 nullable=False, 

52 ) 

53 

54 to_org_id: Mapped[str] = mapped_column( 

55 VARCHAR(length=50), 

56 ForeignKey("organisations.id", onupdate="CASCADE"), 

57 primary_key=True, 

58 nullable=False, 

59 ) 

60 

61 relationship_id: Mapped[int] = mapped_column( 

62 Integer, 

63 ForeignKey("relationship_types.id", ondelete="CASCADE"), 

64 primary_key=True, 

65 nullable=False, 

66 ) 

67 

68 relationship_type = relationship( 

69 RelationshipType, back_populates="edges", uselist=False, lazy="joined" 

70 ) 

71 

72 from_org = relationship( 

73 Organisation, 

74 primaryjoin=from_org_id == Organisation.id, 

75 back_populates="lower_edges", 

76 lazy="joined", 

77 ) 

78 

79 to_org = relationship( 

80 Organisation, 

81 primaryjoin=to_org_id == Organisation.id, 

82 back_populates="higher_edges", 

83 lazy="joined", 

84 ) 

85 

86 @property 

87 def relationship(self) -> str: 

88 return self.relationship_type.name 

89 

90 def __repr__(self) -> str: 

91 return f'<Edge - "{self.from_org_id}" to "{self.to_org_id}">'