Merge pull request #139 from pepkit/history

Project history

pepdbagent/const.py

@@ -20,3 +20,5 @@
 LAST_UPDATE_DATE_KEY = "last_update_date"
 
 PEPHUB_SAMPLE_ID_KEY = "ph_id"
+
+MAX_HISTORY_SAMPLES_NUMBER = 2000

pepdbagent/db_utils.py

@@ -1,10 +1,12 @@
 import datetime
+import enum
 import logging
 from typing import List, Optional
 
 from sqlalchemy import (
     TIMESTAMP,
     BigInteger,
+    Enum,
     FetchedValue,
     ForeignKey,
     Result,
@@ -119,6 +121,10 @@ class Projects(Base):
 
     namespace_mapping: Mapped["User"] = relationship("User", back_populates="projects_mapping")
 
+    history_mapping: Mapped[List["HistoryProjects"]] = relationship(
+        back_populates="project_mapping", cascade="all, delete-orphan"
+    )  # TODO: check if cascade is correct
+
     __table_args__ = (UniqueConstraint("namespace", "name", "tag"),)
 
 
@@ -131,7 +137,6 @@ class Samples(Base):
 
     id: Mapped[int] = mapped_column(primary_key=True)
     sample: Mapped[dict] = mapped_column(JSON, server_default=FetchedValue())
-    row_number: Mapped[int]  # TODO: should be removed
     project_id = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"))
     project_mapping: Mapped["Projects"] = relationship(back_populates="samples_mapping")
     sample_name: Mapped[Optional[str]] = mapped_column()
@@ -245,6 +250,52 @@ class ViewSampleAssociation(Base):
     view: Mapped["Views"] = relationship(back_populates="samples")
 
 
+class HistoryProjects(Base):
+
+    __tablename__ = "project_history"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"))
+    user: Mapped[str] = mapped_column(ForeignKey("users.namespace", ondelete="SET NULL"))
+    update_time: Mapped[datetime.datetime] = mapped_column(
+        TIMESTAMP(timezone=True), default=deliver_update_date
+    )
+    project_yaml: Mapped[dict] = mapped_column(JSON, server_default=FetchedValue())
+
+    project_mapping: Mapped["Projects"] = relationship(
+        "Projects", back_populates="history_mapping"
+    )
+    sample_changes_mapping: Mapped[List["HistorySamples"]] = relationship(
+        back_populates="history_project_mapping", cascade="all, delete-orphan"
+    )
+
+
+class UpdateTypes(enum.Enum):
+    """
+    Enum for the type of update
+    """
+
+    UPDATE = "update"
+    INSERT = "insert"
+    DELETE = "delete"
+
+
+class HistorySamples(Base):
+
+    __tablename__ = "sample_history"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    history_id: Mapped[int] = mapped_column(ForeignKey("project_history.id", ondelete="CASCADE"))
+    guid: Mapped[str] = mapped_column(nullable=False)
+    parent_guid: Mapped[Optional[str]] = mapped_column(nullable=True)
+    sample_json: Mapped[dict] = mapped_column(JSON, server_default=FetchedValue())
+    change_type: Mapped[UpdateTypes] = mapped_column(Enum(UpdateTypes), nullable=False)
+
+    history_project_mapping: Mapped["HistoryProjects"] = relationship(
+        "HistoryProjects", back_populates="sample_changes_mapping"
+    )
+
+
 class BaseEngine:
     """
     A class with base methods, that are used in several classes. e.g. fetch_one or fetch_all

pepdbagent/exceptions.py

@@ -107,3 +107,13 @@ def __init__(self, msg=""):
 class NamespaceNotFoundError(PEPDatabaseAgentError):
     def __init__(self, msg=""):
         super().__init__(f"""Project does not exist. {msg}""")
+
+
+class HistoryNotFoundError(PEPDatabaseAgentError):
+    def __init__(self, msg=""):
+        super().__init__(f"""History does not exist. {msg}""")
+
+
+class UserNotFoundError(PEPDatabaseAgentError):
+    def __init__(self, msg=""):
+        super().__init__(f"""User does not exist. {msg}""")

pepdbagent/models.py

@@ -1,4 +1,5 @@
 # file with pydantic models
+import datetime
 from typing import Dict, List, Optional, Union
 
 from peppy.const import CONFIG_KEY, SAMPLE_RAW_DICT_KEY, SUBSAMPLE_RAW_LIST_KEY
@@ -224,3 +225,24 @@ class NamespaceStats(BaseModel):
     namespace: Union[str, None] = None
     projects_updated: Dict[str, int] = None
     projects_created: Dict[str, int] = None
+
+
+class HistoryChangeModel(BaseModel):
+    """
+    Model for history change
+    """
+
+    change_id: int
+    change_date: datetime.datetime
+    user: str
+
+
+class HistoryAnnotationModel(BaseModel):
+    """
+    History annotation model
+    """
+
+    namespace: str
+    name: str
+    tag: str = DEFAULT_TAG
+    history: List[HistoryChangeModel]