Fixed #73

pepdbagent/db_utils.py

@@ -100,6 +100,22 @@ class Projects(Base):
     stars_mapping: Mapped[List["Stars"]] = relationship(
         back_populates="project_mapping", cascade="all, delete-orphan"
     )
+    # Self-referential relationship. The parent project is the one that was forked to create this one.
+    forked_from_id: Mapped[Optional[int]] = mapped_column(
+        ForeignKey("projects.id", ondelete="SET NULL"), nullable=True
+    )
+    forked_from_mapping = relationship(
+        "Projects",
+        back_populates="forked_to_mapping",
+        remote_side=[id],
+        single_parent=True,
+        cascade="all",
+    )
+
+    forked_to_mapping = relationship(
+        "Projects", back_populates="forked_from_mapping", cascade="all"
+    )
+
     __table_args__ = (UniqueConstraint("namespace", "name", "tag"),)
 
 

pepdbagent/models.py

@@ -22,6 +22,7 @@ class AnnotationModel(BaseModel):
     pep_schema: Optional[str]
     pop: Optional[bool] = False
     stars_number: Optional[int] = 0
+    forked_from: Optional[Union[str, None]] = None
 
     model_config = ConfigDict(
         validate_assignment=True,

pepdbagent/modules/annotation.py

@@ -214,6 +214,9 @@ def _get_single_annotation(
                     pep_schema=query_result.pep_schema,
                     pop=query_result.pop,
                     stars_number=len(query_result.stars_mapping),
+                    forked_from=f"{query_result.forked_from_mapping.namespace}/{query_result.forked_from_mapping.name}:{query_result.forked_from_mapping.tag}"
+                    if query_result.forked_from_id
+                    else None,
                 )
                 _LOGGER.info(
                     f"Annotation of the project '{namespace}/{name}:{tag}' has been found!"
@@ -343,6 +346,9 @@ def _get_projects(
                         pep_schema=result.pep_schema,
                         pop=result.pop,
                         stars_number=len(result.stars_mapping),
+                        forked_from=f"{result.forked_from_mapping.namespace}/{result.forked_from_mapping.name}:{result.forked_from_mapping.tag}"
+                        if result.forked_from_id
+                        else None,
                     )
                 )
         return results_list
@@ -547,6 +553,9 @@ def get_by_rp_list(
                         pep_schema=project_obj.pep_schema,
                         pop=project_obj.pop,
                         stars_number=len(project_obj.stars_mapping),
+                        forked_from=f"{project_obj.forked_from_mapping.namespace}/{project_obj.forked_from_mapping.name}:{project_obj.forked_from_mapping.tag}"
+                        if project_obj.forked_from_mapping
+                        else None,
                     )
                     anno_results.append(annot)
 

pepdbagent/modules/project.py

@@ -624,3 +624,44 @@ def get_project_id(self, namespace: str, name: str, tag: str) -> Union[int, None
         if result:
             return result[0]
         return None
+
+    def fork(
+        self,
+        original_namespace: str,
+        original_name: str,
+        original_tag: str,
+        fork_namespace: str,
+        fork_name: str = None,
+        fork_tag: str = None,
+    ):
+        """
+
+        :param original_namespace:
+        :param original_name:
+        :param original_tag:
+        :param fork_namespace:
+        :param fork_name:
+        :param fork_tag:
+        :return:
+        """
+        self.create(
+            project=self.get(
+                namespace=original_namespace,
+                name=original_name,
+                tag=original_tag,
+            ),
+            namespace=fork_namespace,
+            name=fork_name,
+            tag=fork_tag,
+        )
+        original_id = self.get_project_id(original_namespace, original_name, original_tag)
+        with Session(self._sa_engine) as session:
+            statement = select(Projects).where(
+                Projects.namespace == fork_namespace,
+                Projects.name == fork_name,
+                Projects.tag == fork_tag,
+            )
+            prj = session.scalar(statement)
+            prj.forked_from_id = original_id
+            session.commit()
+        return None

tests/test_pepagent.py

@@ -124,6 +124,128 @@ def test_delete_project(self, initiate_pepdb_con, namespace, name):
         with pytest.raises(ProjectNotFoundError, match="Project does not exist."):
             initiate_pepdb_con.project.get(namespace=namespace, name=name, tag="default")
 
+    @pytest.mark.parametrize(
+        "namespace, name",
+        [
+            ["namespace1", "amendments1"],
+            ["namespace1", "amendments2"],
+            ["namespace2", "derive"],
+            ["namespace2", "imply"],
+        ],
+    )
+    def test_fork_projects(self, initiate_pepdb_con, namespace, name):
+        initiate_pepdb_con.project.fork(
+            original_namespace=namespace,
+            original_name=name,
+            original_tag="default",
+            fork_namespace="new_namespace",
+            fork_name="new_name",
+            fork_tag="new_tag",
+        )
+
+        assert initiate_pepdb_con.project.exists(
+            namespace="new_namespace", name="new_name", tag="new_tag"
+        )
+        result = initiate_pepdb_con.annotation.get(
+            namespace="new_namespace", name="new_name", tag="new_tag"
+        )
+        assert result.results[0].forked_from == f"{namespace}/{name}:default"
+
+    @pytest.mark.parametrize(
+        "namespace, name",
+        [
+            ["namespace1", "amendments1"],
+            ["namespace1", "amendments2"],
+        ],
+    )
+    def test_parent_project_delete(self, initiate_pepdb_con, namespace, name):
+        """
+        Test if parent project is deleted, forked project is not deleted
+        """
+        initiate_pepdb_con.project.fork(
+            original_namespace=namespace,
+            original_name=name,
+            original_tag="default",
+            fork_namespace="new_namespace",
+            fork_name="new_name",
+            fork_tag="new_tag",
+        )
+
+        assert initiate_pepdb_con.project.exists(
+            namespace="new_namespace", name="new_name", tag="new_tag"
+        )
+        initiate_pepdb_con.project.delete(namespace=namespace, name=name, tag="default")
+        assert initiate_pepdb_con.project.exists(
+            namespace="new_namespace", name="new_name", tag="new_tag"
+        )
+
+    @pytest.mark.parametrize(
+        "namespace, name",
+        [
+            ["namespace1", "amendments1"],
+            ["namespace1", "amendments2"],
+        ],
+    )
+    def test_child_project_delete(self, initiate_pepdb_con, namespace, name):
+        """
+        Test if child project is deleted, parent project is not deleted
+        """
+        initiate_pepdb_con.project.fork(
+            original_namespace=namespace,
+            original_name=name,
+            original_tag="default",
+            fork_namespace="new_namespace",
+            fork_name="new_name",
+            fork_tag="new_tag",
+        )
+
+        assert initiate_pepdb_con.project.exists(
+            namespace="new_namespace", name="new_name", tag="new_tag"
+        )
+        assert initiate_pepdb_con.project.exists(namespace=namespace, name=name, tag="default")
+        initiate_pepdb_con.project.delete(
+            namespace="new_namespace", name="new_name", tag="new_tag"
+        )
+        assert initiate_pepdb_con.project.exists(namespace=namespace, name=name, tag="default")
+
+    @pytest.mark.parametrize(
+        "namespace, name",
+        [
+            ["namespace1", "amendments1"],
+            ["namespace1", "amendments2"],
+        ],
+    )
+    def test_project_can_be_forked_twice(self, initiate_pepdb_con, namespace, name):
+        """
+        Test if project can be forked twice
+        """
+        initiate_pepdb_con.project.fork(
+            original_namespace=namespace,
+            original_name=name,
+            original_tag="default",
+            fork_namespace="new_namespace",
+            fork_name="new_name",
+            fork_tag="new_tag",
+        )
+        initiate_pepdb_con.project.fork(
+            original_namespace=namespace,
+            original_name=name,
+            original_tag="default",
+            fork_namespace="new_namespace2",
+            fork_name="new_name2",
+            fork_tag="new_tag2",
+        )
+
+        result = initiate_pepdb_con.annotation.get(
+            namespace="new_namespace", name="new_name", tag="new_tag"
+        )
+        assert result.results[0].forked_from == f"{namespace}/{name}:default"
+
+        result = initiate_pepdb_con.annotation.get(
+            namespace="new_namespace2", name="new_name2", tag="new_tag2"
+        )
+        assert result.results[0].forked_from == f"{namespace}/{name}:default"
+
 
 @pytest.mark.skipif(
     not db_setup(),
@@ -445,6 +567,7 @@ def test_all_annotations_are_returned(self, initiate_pepdb_con, namespace, name)
             "pep_schema",
             "pop",
             "stars_number",
+            "forked_from",
         }
 
     @pytest.mark.parametrize(