Coverage for postrfp / model / datafeeds.py: 100%

1from typing import Optional, TYPE_CHECKING 

2 

3import sqids 

4from sqlalchemy import ( 

5 ForeignKey, 

6 VARCHAR, 

7) 

8from sqlalchemy.orm import Mapped, mapped_column, relationship 

9 

10 

11from postrfp.model.meta import Base, HTTPHeadersMixin, SqidsMixin 

12 

13if TYPE_CHECKING: 

14 from postrfp.model.humans import Organisation 

15 from postrfp.model.jobs import JobExecution 

16 

17 

18SQID = sqids.Sqids( 

19 alphabet="qu42jfwe8apgxz9d7shym5ckvb6nrt3", 

20 min_length=5, 

21) 

22 

23 

24class Datafeed(Base, HTTPHeadersMixin, SqidsMixin): 

25 """ 

26 Configuration for fetching external data and mapping it to Content items. 

27 """ 

28 

29 __tablename__ = "datafeeds" 

30 _sqids_alphabet = "qu42jfwe8apgxz9d7shym5ckvb6nrt3" 

31 

32 org_id: Mapped[str] = mapped_column( 

33 VARCHAR(length=50), 

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

35 nullable=False, 

36 index=True, 

37 ) 

38 

39 name: Mapped[str] = mapped_column(VARCHAR(length=128), nullable=False) 

40 

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

42 VARCHAR(length=512), nullable=True 

43 ) 

44 

45 source_url: Mapped[str] = mapped_column( 

46 VARCHAR(length=512), 

47 nullable=False, 

48 comment="External URL to fetch data from. Supports templating (e.g. {project_id})", 

49 ) 

50 

51 transform_expression: Mapped[Optional[str]] = mapped_column( 

52 VARCHAR(length=4000), 

53 nullable=True, 

54 comment="CEL expression to transform external JSON to Content format", 

55 ) 

56 

57 # Relationships 

58 organisation: Mapped["Organisation"] = relationship("Organisation") 

59 

60 executions: Mapped[list["JobExecution"]] = relationship( 

61 "JobExecution", back_populates="datafeed", cascade="all, delete-orphan" 

62 ) 

63 

64 def __repr__(self) -> str: 

65 return f"<Datafeed {self.name} ({self.id})>"