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

1from typing import Optional 

2 

3from sqlalchemy import ForeignKey 

4from sqlalchemy.orm import relationship, Mapped, mapped_column 

5from sqlalchemy.dialects import mysql 

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 mysql.VARCHAR(length=50), 

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

18 nullable=True, 

19 ) 

20 name: Mapped[Optional[str]] = mapped_column(mysql.VARCHAR(length=128), nullable=True) 

21 description: Mapped[Optional[str]] = mapped_column(mysql.VARCHAR(length=256), nullable=True) 

22 

23 organisation = relationship( 

24 Organisation, 

25 primaryjoin=org_id == Organisation.id, 

26 back_populates="relationship_types", 

27 ) 

28 

29 edges = relationship( 

30 "Edge", 

31 back_populates="relationship_type", 

32 cascade="all, delete", 

33 passive_deletes=True, 

34 ) 

35 

36 def __repr__(self) -> str: 

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

38 

39 

40class Edge(Base): 

41 __tablename__ = "edges" 

42 

43 id = None # type: ignore 

44 

45 from_org_id: Mapped[str] = mapped_column( 

46 mysql.VARCHAR(length=50), 

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

48 primary_key=True, 

49 nullable=False, 

50 ) 

51 

52 to_org_id: Mapped[str] = mapped_column( 

53 mysql.VARCHAR(length=50), 

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

55 primary_key=True, 

56 nullable=False, 

57 ) 

58 

59 relationship_id: Mapped[int] = mapped_column( 

60 Integer, 

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

62 primary_key=True, 

63 nullable=False, 

64 ) 

65 

66 relationship_type = relationship( 

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

68 ) 

69 

70 from_org = relationship( 

71 Organisation, 

72 primaryjoin=from_org_id == Organisation.id, 

73 back_populates="lower_edges", 

74 lazy="joined", 

75 ) 

76 

77 to_org = relationship( 

78 Organisation, 

79 primaryjoin=to_org_id == Organisation.id, 

80 back_populates="higher_edges", 

81 lazy="joined", 

82 ) 

83 

84 @property 

85 def relationship(self) -> str: 

86 return self.relationship_type.name 

87 

88 def __repr__(self) -> str: 

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