104 favorites

pepdbagent/_version.py

@@ -1 +1 @@
-__version__ = "0.7.0a1"
+__version__ = "0.7.0a2"

pepdbagent/db_utils.py

@@ -97,7 +97,9 @@ class Projects(Base):
     subsamples_mapping: Mapped[List["Subsamples"]] = relationship(
         back_populates="subsample_mapping", cascade="all, delete-orphan"
     )
-
+    stars_mapping: Mapped[List["Stars"]] = relationship(
+        back_populates="project_mapping", cascade="all, delete-orphan"
+    )
     __table_args__ = (UniqueConstraint("namespace", "name", "tag"),)
 
 
@@ -130,6 +132,33 @@ class Subsamples(Base):
     subsample_mapping: Mapped["Projects"] = relationship(back_populates="subsamples_mapping")
 
 
+class User(Base):
+    """
+    User table representation in the database
+    """
+
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    namespace: Mapped[str]
+    stars_mapping: Mapped[List["Stars"]] = relationship(
+        back_populates="user_mapping", cascade="all, delete-orphan"
+    )
+
+
+class Stars(Base):
+    """
+    FavoriteProjects table representation in the database
+    """
+
+    __tablename__ = "stars"
+
+    user_id = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), primary_key=True)
+    project_id = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), primary_key=True)
+    user_mapping: Mapped[List["User"]] = relationship(back_populates="stars_mapping")
+    project_mapping: Mapped["Projects"] = relationship(back_populates="stars_mapping")
+
+
 class BaseEngine:
     """
     A class with base methods, that are used in several classes. e.g. fetch_one or fetch_all

pepdbagent/exceptions.py

@@ -36,3 +36,21 @@ def __init__(self, msg=""):
 class FilterError(PEPDatabaseAgentError):
     def __init__(self, msg=""):
         super().__init__(f"""pepdbagent filter error. {msg}""")
+
+
+class ProjectNotInFavorites(PEPDatabaseAgentError):
+    """
+    Project doesn't exist in favorites
+    """
+
+    def __init__(self, msg=""):
+        super().__init__(f"""Project is not in favorites list. {msg}""")
+
+
+class ProjectAlreadyInFavorites(PEPDatabaseAgentError):
+    """
+    Project doesn't exist in favorites
+    """
+
+    def __init__(self, msg=""):
+        super().__init__(f"""Project is already in favorites list. {msg}""")

pepdbagent/models.py

@@ -1,6 +1,8 @@
 # file with pydantic models
 from typing import List, Optional, Union
-from pydantic import BaseModel, Extra, Field, validator, ConfigDict, field_validator
+from pydantic import BaseModel, Field, ConfigDict, field_validator
+
+from pepdbagent.const import DEFAULT_TAG
 
 
 class AnnotationModel(BaseModel):
@@ -19,6 +21,7 @@ class AnnotationModel(BaseModel):
     digest: Optional[str]
     pep_schema: Optional[str]
     pop: Optional[bool] = False
+    stars_number: Optional[int] = 0
 
     model_config = ConfigDict(
         validate_assignment=True,
@@ -146,3 +149,13 @@ class ListOfNamespaceInfo(BaseModel):
     number_of_namespaces: int
     limit: int
     results: List[NamespaceInfo]
+
+
+class ProjectRegistryPath(BaseModel):
+    """
+    Project Namespace
+    """
+
+    namespace: str
+    name: str
+    tag: str = DEFAULT_TAG

pepdbagent/modules/annotation.py

@@ -4,6 +4,7 @@
 
 from sqlalchemy import and_, func, or_, select
 from sqlalchemy.sql.selectable import Select
+from sqlalchemy.orm import Session
 
 from pepdbagent.const import (
     DEFAULT_LIMIT,
@@ -177,19 +178,7 @@ def _get_single_annotation(
         _LOGGER.info(f"Getting annotation of the project: '{namespace}/{name}:{tag}'")
         admin_tuple = tuple_converter(admin)
 
-        statement = select(
-            Projects.namespace,
-            Projects.name,
-            Projects.tag,
-            Projects.private,
-            Projects.description,
-            Projects.number_of_samples,
-            Projects.submission_date,
-            Projects.last_update_date,
-            Projects.digest,
-            Projects.pep_schema,
-            Projects.pop,
-        ).where(
+        statement = select(Projects).where(
             and_(
                 Projects.name == name,
                 Projects.namespace == namespace,
@@ -200,26 +189,30 @@ def _get_single_annotation(
                 ),
             )
         )
-        query_result = self._pep_db_engine.session_execute(statement).first()
-
-        if query_result:
-            annot = AnnotationModel(
-                namespace=query_result.namespace,
-                name=query_result.name,
-                tag=query_result.tag,
-                is_private=query_result.private,
-                description=query_result.description,
-                number_of_samples=query_result.number_of_samples,
-                submission_date=str(query_result.submission_date),
-                last_update_date=str(query_result.last_update_date),
-                digest=query_result.digest,
-                pep_schema=query_result.pep_schema,
-                pop=query_result.pop,
-            )
-            _LOGGER.info(f"Annotation of the project '{namespace}/{name}:{tag}' has been found!")
-            return annot
-        else:
-            raise ProjectNotFoundError(f"Project '{namespace}/{name}:{tag}' was not found.")
+        with Session(self._sa_engine) as session:
+            query_result = session.scalar(statement)
+
+            if query_result:
+                annot = AnnotationModel(
+                    namespace=query_result.namespace,
+                    name=query_result.name,
+                    tag=query_result.tag,
+                    is_private=query_result.private,
+                    description=query_result.description,
+                    number_of_samples=query_result.number_of_samples,
+                    submission_date=str(query_result.submission_date),
+                    last_update_date=str(query_result.last_update_date),
+                    digest=query_result.digest,
+                    pep_schema=query_result.pep_schema,
+                    pop=query_result.pop,
+                    stars_number=len(query_result.stars_mapping),
+                )
+                _LOGGER.info(
+                    f"Annotation of the project '{namespace}/{name}:{tag}' has been found!"
+                )
+                return annot
+            else:
+                raise ProjectNotFoundError(f"Project '{namespace}/{name}:{tag}' was not found.")
 
     def _count_projects(
         self,
@@ -297,19 +290,7 @@ def _get_projects(
 
         if admin is None:
             admin = []
-        statement = select(
-            Projects.namespace,
-            Projects.name,
-            Projects.tag,
-            Projects.private,
-            Projects.description,
-            Projects.number_of_samples,
-            Projects.submission_date,
-            Projects.last_update_date,
-            Projects.digest,
-            Projects.pep_schema,
-            Projects.pop,
-        ).select_from(Projects)
+        statement = select(Projects.id)
 
         statement = self._add_condition(
             statement,
@@ -323,25 +304,29 @@ def _get_projects(
         statement = self._add_order_by_keyword(statement, by=order_by, desc=order_desc)
         statement = statement.limit(limit).offset(offset)
 
-        query_results = self._pep_db_engine.session_execute(statement).all()
+        id_results = self._pep_db_engine.session_execute(statement).all()
 
         results_list = []
-        for result in query_results:
-            results_list.append(
-                AnnotationModel(
-                    namespace=result.namespace,
-                    name=result.name,
-                    tag=result.tag,
-                    is_private=result.private,
-                    description=result.description,
-                    number_of_samples=result.number_of_samples,
-                    submission_date=str(result.submission_date),
-                    last_update_date=str(result.last_update_date),
-                    digest=result.digest,
-                    pep_schema=result.pep_schema,
-                    pop=result.pop,
+        with Session(self._sa_engine) as session:
+            for prj_ids in id_results:
+                result = session.scalar(select(Projects).where(Projects.id == prj_ids[0]))
+
+                results_list.append(
+                    AnnotationModel(
+                        namespace=result.namespace,
+                        name=result.name,
+                        tag=result.tag,
+                        is_private=result.private,
+                        description=result.description,
+                        number_of_samples=result.number_of_samples,
+                        submission_date=str(result.submission_date),
+                        last_update_date=str(result.last_update_date),
+                        digest=result.digest,
+                        pep_schema=result.pep_schema,
+                        pop=result.pop,
+                        stars_number=len(result.stars_mapping),
+                    )
                 )
-            )
         return results_list
 
     @staticmethod

pepdbagent/modules/namespace.py

@@ -1,5 +1,5 @@
 import logging
-from typing import List, Union
+from typing import List, Union, Tuple
 
 from sqlalchemy import distinct, func, or_, select, text
 from sqlalchemy.sql.selectable import Select
@@ -141,7 +141,7 @@ def _count_namespace(self, search_str: str = None, admin_nsp: tuple = tuple()) -
     def _add_condition(
         statement: Select,
         search_str: str = None,
-        admin_list: Union[str, List[str]] = None,
+        admin_list: Union[Tuple[str], List[str], str] = None,
     ) -> Select:
         """
         Add where clause to sqlalchemy statement (in namespace search)

pepdbagent/modules/project.py

@@ -605,3 +605,22 @@ def _add_subsamples_to_project(
                 projects_sa.subsamples_mapping.append(
                     Subsamples(subsample=sub_item, subsample_number=i, row_number=row_number)
                 )
+
+    def get_project_id(self, namespace: str, name: str, tag: str) -> Union[int, None]:
+        """
+        Get Project id by providing namespace, name, and tag
+
+        :param namespace: project namespace
+        :param name: project name
+        :param tag: project tag
+        :return: projects id
+        """
+        statement = select(Projects.id).where(
+            and_(Projects.namespace == namespace, Projects.name == name, Projects.tag == tag)
+        )
+        with Session(self._sa_engine) as session:
+            result = session.execute(statement).one_or_none()
+
+        if result:
+            return result[0]
+        return None