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

1from sqlalchemy import Column, ForeignKey 

2from sqlalchemy.orm import relationship, backref 

3from sqlalchemy.dialects import mysql 

4from sqlalchemy.sql.sqltypes import Integer 

5 

6from .humans import Organisation 

7from .meta import Base 

8 

9 

10class RelationshipType(Base): 

11 

12 __tablename__ = "relationship_types" 

13 

14 org_id = Column( 

15 mysql.VARCHAR(length=50), 

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

17 ) 

18 name = Column(mysql.VARCHAR(length=128)) 

19 description = Column(mysql.VARCHAR(length=256)) 

20 

21 organisation = relationship( 

22 Organisation, 

23 primaryjoin=org_id == Organisation.id, 

24 backref=backref("relationship_types", lazy="dynamic", cascade="all,delete"), 

25 ) 

26 

27 edges = relationship( 

28 "Edge", 

29 back_populates="relationship_type", 

30 cascade="all, delete", 

31 passive_deletes=True, 

32 ) 

33 

34 def __repr__(self): 

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

36 

37 

38class Edge(Base): 

39 

40 __tablename__ = "edges" 

41 

42 id = None # Override base class ID 

43 

44 from_org_id = Column( 

45 mysql.VARCHAR(length=50), 

46 ForeignKey("organisations.id", onupdate='CASCADE'), 

47 primary_key=True 

48 ) 

49 

50 to_org_id = Column( 

51 mysql.VARCHAR(length=50), 

52 ForeignKey("organisations.id", onupdate='CASCADE'), 

53 primary_key=True 

54 ) 

55 

56 relationship_id = Column( 

57 Integer, 

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

59 primary_key=True, 

60 ) 

61 

62 relationship_type = relationship( 

63 RelationshipType, back_populates="edges", uselist=False, lazy='joined' 

64 ) 

65 

66 from_org = relationship( 

67 Organisation, 

68 primaryjoin=from_org_id == Organisation.id, 

69 backref=backref("lower_edges", cascade="all, delete"), 

70 lazy='joined' 

71 ) 

72 

73 to_org = relationship( 

74 Organisation, 

75 primaryjoin=to_org_id == Organisation.id, 

76 backref=backref("higher_edges", cascade="all,delete"), 

77 lazy='joined' 

78 ) 

79 

80 @property 

81 def relationship(self): 

82 return self.relationship_type.name 

83 

84 def __repr__(self): 

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