Catalog returns only the functions that you can run (#1540)

* move program model access to repository

* remove unused code

* fix linter

* fixed tests

* update swagger information

* added logger in the repository class

* rename provider method

* added TypeFilter enum

* fix typo

gateway/api/repositories/programs.py

@@ -0,0 +1,201 @@
+"""
+Repository implementatio for Programs model
+"""
+import logging
+
+from typing import Any, List
+
+from django.db.models import Q
+from django.contrib.auth.models import Group, Permission
+
+from api.models import RUN_PROGRAM_PERMISSION, VIEW_PROGRAM_PERMISSION, Program
+
+
+logger = logging.getLogger("gateway")
+
+
+class ProgramRepository:
+    """
+    The main objective of this class is to manage the access to the model
+    """
+
+    def get_functions(self, author) -> List[Program] | Any:
+        """
+        Returns all the functions available to the user. This means:
+          - User functions where the user is the author
+          - Provider functions with view permissions
+
+        Args:
+            author: Django author from who retrieve the functions
+
+        Returns:
+            List[Program] | Any: all the functions available to the user
+        """
+
+        view_program_permission = Permission.objects.get(
+            codename=VIEW_PROGRAM_PERMISSION
+        )
+
+        user_criteria = Q(user=author)
+        view_permission_criteria = Q(permissions=view_program_permission)
+
+        author_groups_with_view_permissions = Group.objects.filter(
+            user_criteria & view_permission_criteria
+        )
+
+        author_criteria = Q(author=author)
+        author_groups_with_view_permissions_criteria = Q(
+            instances__in=author_groups_with_view_permissions
+        )
+
+        result_queryset = Program.objects.filter(
+            author_criteria | author_groups_with_view_permissions_criteria
+        ).distinct()
+
+        count = result_queryset.count()
+        logger.info("[%d] Functions found for author [%s]", count, author.id)
+
+        return result_queryset
+
+    def get_user_functions(self, author) -> List[Program] | Any:
+        """
+        Returns the user functions available to the user. This means:
+          - User functions where the user is the author
+          - Provider is None
+
+        Args:
+            author: Django author from who retrieve the functions
+
+        Returns:
+            List[Program] | Any: user functions available to the user
+        """
+
+        author_criteria = Q(author=author)
+        provider_criteria = Q(provider=None)
+
+        result_queryset = Program.objects.filter(
+            author_criteria & provider_criteria
+        ).distinct()
+
+        count = result_queryset.count()
+        logger.info("[%d] user Functions found for author [%s]", count, author.id)
+
+        return result_queryset
+
+    def get_provider_functions_with_run_permissions(
+        self, author
+    ) -> List[Program] | Any:
+        """
+        Returns the provider functions available to the user. This means:
+          - Provider functions where the user has run permissions
+          - Provider is NOT None
+
+        Args:
+            author: Django author from who retrieve the functions
+
+        Returns:
+            List[Program] | Any: providers functions available to the user
+        """
+
+        run_program_permission = Permission.objects.get(codename=RUN_PROGRAM_PERMISSION)
+
+        user_criteria = Q(user=author)
+        run_permission_criteria = Q(permissions=run_program_permission)
+        author_groups_with_run_permissions = Group.objects.filter(
+            user_criteria & run_permission_criteria
+        )
+
+        author_groups_with_run_permissions_criteria = Q(
+            instances__in=author_groups_with_run_permissions
+        )
+
+        provider_exists_criteria = ~Q(provider=None)
+
+        result_queryset = Program.objects.filter(
+            author_groups_with_run_permissions_criteria & provider_exists_criteria
+        ).distinct()
+
+        count = result_queryset.count()
+        logger.info("[%d] provider Functions found for author [%s]", count, author.id)
+
+        return result_queryset
+
+    def get_user_function_by_title(self, author, title: str) -> Program | Any:
+        """
+        Returns the user function associated to a title:
+
+        Args:
+            author: Django author from who retrieve the function
+            title: Title that the function must have to find it
+
+        Returns:
+            Program | Any: user function with the specific title
+        """
+
+        author_criteria = Q(author=author)
+        title_criteria = Q(title=title)
+
+        result_queryset = Program.objects.filter(
+            author_criteria & title_criteria
+        ).first()
+
+        if result_queryset is None:
+            logger.warning(
+                "Function [%s] was not found or author [%s] doesn't have access to it",
+                title,
+                author.id,
+            )
+
+        return result_queryset
+
+    def get_provider_function_by_title(
+        self, author, title: str, provider_name: str
+    ) -> Program | Any:
+        """
+        Returns the provider function associated to:
+          - A Function title
+          - A Provider
+          - Author must have view permission to see it or be the author
+
+        Args:
+            author: Django author from who retrieve the function
+            title: Title that the function must have to find it
+            provider: Provider associated to the function
+
+        Returns:
+            Program | Any: provider function with the specific
+                title and provider
+        """
+
+        view_program_permission = Permission.objects.get(
+            codename=VIEW_PROGRAM_PERMISSION
+        )
+
+        user_criteria = Q(user=author)
+        view_permission_criteria = Q(permissions=view_program_permission)
+
+        author_groups_with_view_permissions = Group.objects.filter(
+            user_criteria & view_permission_criteria
+        )
+
+        author_criteria = Q(author=author)
+        author_groups_with_view_permissions_criteria = Q(
+            instances__in=author_groups_with_view_permissions
+        )
+
+        title_criteria = Q(title=title, provider__name=provider_name)
+
+        result_queryset = Program.objects.filter(
+            (author_criteria | author_groups_with_view_permissions_criteria)
+            & title_criteria
+        ).first()
+
+        if result_queryset is None:
+            logger.warning(
+                "Function [%s/%s] was not found or author [%s] doesn't have access to it",
+                provider_name,
+                title,
+                author.id,
+            )
+
+        return result_queryset

gateway/api/v1/views/programs.py

@@ -2,7 +2,7 @@
 Programs view api for V1.
 """
 
-# pylint: disable=duplicate-code
+from drf_yasg import openapi
 from drf_yasg.utils import swagger_auto_schema
 from rest_framework import permissions, status
 from rest_framework.decorators import action
@@ -42,6 +42,15 @@ def get_serializer_job(*args, **kwargs):
 
     @swagger_auto_schema(
         operation_description="List author Qiskit Functions",
+        manual_parameters=[
+            openapi.Parameter(
+                "filter",
+                openapi.IN_QUERY,
+                description="Filters that you can apply for list: serverless, catalog or empty",
+                type=openapi.TYPE_STRING,
+                required=False,
+            ),
+        ],
         responses={status.HTTP_200_OK: v1_serializers.ProgramSerializer(many=True)},
     )
     def list(self, request):
@@ -64,3 +73,26 @@ def upload(self, request):
     @action(methods=["POST"], detail=False)
     def run(self, request):
         return super().run(request)
+
+    @swagger_auto_schema(
+        operation_description="Retrieve a Qiskit Function using the title",
+        manual_parameters=[
+            openapi.Parameter(
+                "title",
+                openapi.IN_PATH,
+                description="The title of the function",
+                type=openapi.TYPE_STRING,
+            ),
+            openapi.Parameter(
+                "provider",
+                openapi.IN_QUERY,
+                description="The provider in case the function is owned by a provider",
+                type=openapi.TYPE_STRING,
+                required=False,
+            ),
+        ],
+        responses={status.HTTP_200_OK: v1_serializers.ProgramSerializer},
+    )
+    @action(methods=["GET"], detail=False, url_path="get_by_title/(?P<title>[^/.]+)")
+    def get_by_title(self, request, title):
+        return super().get_by_title(request, title)

gateway/api/views/enums/type_filter.py

@@ -0,0 +1,16 @@
+"""
+This class defines TypeFilter enum for views:
+"""
+
+from enum import Enum
+
+
+class TypeFilter(str, Enum):
+    """
+    TypeFilter values for the view end-points:
+    - SERVERLESS
+    - CATALOG
+    """
+
+    SERVERLESS = "serverless"
+    CATALOG = "catalog"

gateway/api/views/jobs.py

@@ -25,6 +25,7 @@
 from qiskit_ibm_runtime import RuntimeInvalidStateError, QiskitRuntimeService
 from api.models import Job, RuntimeJob
 from api.ray import get_job_handler
+from api.views.enums.type_filter import TypeFilter
 
 # pylint: disable=duplicate-code
 logger = logging.getLogger("gateway")
@@ -56,13 +57,13 @@ def get_serializer_class(self):
     def get_queryset(self):
         type_filter = self.request.query_params.get("filter")
         if type_filter:
-            if type_filter == "catalog":
+            if type_filter == TypeFilter.CATALOG:
                 user_criteria = Q(author=self.request.user)
                 provider_exists_criteria = ~Q(program__provider=None)
                 return Job.objects.filter(
                     user_criteria & provider_exists_criteria
                 ).order_by("-created")
-            if type_filter == "serverless":
+            if type_filter == TypeFilter.SERVERLESS:
                 user_criteria = Q(author=self.request.user)
                 provider_not_exists_criteria = Q(program__provider=None)
                 return Job.objects.filter(