Coverage for src/couchers/models/activeness_probe.py: 100%

1import enum 

2from datetime import datetime 

3from typing import TYPE_CHECKING 

4 

5from sqlalchemy import BigInteger, CheckConstraint, DateTime, Enum, ForeignKey, Index, Integer, func 

6from sqlalchemy.ext.hybrid import hybrid_property 

7from sqlalchemy.orm import Mapped, mapped_column, relationship 

8 

9from couchers.models.base import Base 

10 

11if TYPE_CHECKING: 

12 from couchers.models.users import User 

13 

14 

15class ActivenessProbeStatus(enum.Enum): 

16 # no response yet 

17 pending = enum.auto() 

18 

19 # didn't respond on time 

20 expired = enum.auto() 

21 

22 # responded that they're still active 

23 still_active = enum.auto() 

24 

25 # responded that they're no longer active 

26 no_longer_active = enum.auto() 

27 

28 

29class ActivenessProbe(Base): 

30 """ 

31 Activeness probes are used to gauge if users are still active: we send them a notification and ask them to respond, 

32 we use this data both to help indicate response rate, as well as to make sure only those who are actively hosting 

33 show up as such. 

34 """ 

35 

36 __tablename__ = "activeness_probes" 

37 

38 id: Mapped[int] = mapped_column(BigInteger, primary_key=True) 

39 

40 user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) 

41 # the time this probe was initiated 

42 probe_initiated: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) 

43 # the number of reminders sent for this probe 

44 notifications_sent: Mapped[int] = mapped_column(Integer, server_default="0") 

45 

46 # the time of response 

47 responded: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None) 

48 # the response value 

49 response: Mapped[ActivenessProbeStatus] = mapped_column( 

50 Enum(ActivenessProbeStatus), default=ActivenessProbeStatus.pending 

51 ) 

52 

53 @hybrid_property 

54 def is_pending(self) -> bool: 

55 return self.responded == None 

56 

57 user: Mapped["User"] = relationship("User", back_populates="pending_activeness_probe") 

58 

59 __table_args__ = ( 

60 # a user can have at most one pending activeness probe at a time 

61 Index( 

62 "ix_activeness_probe_unique_pending_response", 

63 user_id, 

64 unique=True, 

65 postgresql_where=responded.is_(None), 

66 ), 

67 # response time is none iff response is pending 

68 CheckConstraint( 

69 "(responded IS NULL AND response = 'pending') OR (responded IS NOT NULL AND response != 'pending')", 

70 name="pending_has_no_responded", 

71 ), 

72 )