Improve get jobs and get job logic with new search parameters

client/qiskit_serverless/core/clients/serverless_client.py

@@ -192,10 +192,29 @@ def dependencies_versions(self):
 
     @_trace_job("list")
     def jobs(self, **kwargs) -> List[Job]:
+        """Retrieve a list of jobs with optional filtering.
+
+        Args:
+            limit (int, optional): Maximum number of jobs to return. Defaults to 10.
+            offset (int, optional): Number of jobs to skip. Defaults to 0.
+            status (str, optional): Filter by job status.
+            created_after (str, optional): Filter jobs created after this timestamp.
+            function_name (str, optional): Filter by function name.
+            **kwargs: Additional query parameters.
+
+        Returns:
+            List[Job]: List of Job objects matching the criteria.
+        """
         limit = kwargs.get("limit", 10)
         kwargs["limit"] = limit
         offset = kwargs.get("offset", 0)
         kwargs["offset"] = offset
+        status = kwargs.get("status", None)
+        kwargs["status"] = status
+        created_after = kwargs.get("created_after", None)
+        kwargs["created_after"] = created_after
+        function_name = kwargs.get("function_name", None)
+        kwargs["function"] = function_name
 
         response_data = safe_json_request_as_dict(
             request=lambda: requests.get(
@@ -215,19 +234,22 @@ def jobs(self, **kwargs) -> List[Job]:
 
     @_trace_job("provider_list")
     def provider_jobs(self, function: QiskitFunction, **kwargs) -> List[Job]:
-        """List of jobs created in this provider and function.
+        """Retrieve jobs for a specific provider and function.
 
         Args:
-            function: QiskitFunction
-            **kwargs: additional parameters for the request
-
-        Raises:
-            QiskitServerlessException: validation exception
+            function (QiskitFunction): Function object.
+            limit (int, optional): Maximum number of jobs to return. Defaults to 10.
+            offset (int, optional): Number of jobs to skip. Defaults to 0.
+            status (str, optional): Filter by job status.
+            created_after (str, optional): Filter jobs created after this timestamp.
+            **kwargs: Additional query parameters.
 
         Returns:
-            [Job] : list of jobs
-        """
+            List[Job]: List of Job objects for the specified provider and function.
 
+        Raises:
+            QiskitServerlessException: If the function doesn't have an associated provider.
+        """
         if not function.provider:
             raise QiskitServerlessException("`function` doesn't have a provider.")
 
@@ -237,6 +259,10 @@ def provider_jobs(self, function: QiskitFunction, **kwargs) -> List[Job]:
         kwargs["offset"] = offset
         kwargs["function"] = function.title
         kwargs["provider"] = function.provider
+        status = kwargs.get("status", None)
+        kwargs["status"] = status
+        created_after = kwargs.get("created_after", None)
+        kwargs["created_after"] = created_after
 
         response_data = safe_json_request_as_dict(
             request=lambda: requests.get(

gateway/api/repositories/functions.py

@@ -222,6 +222,40 @@ def get_function_by_permission(
 
         return self.get_user_function(author=user, title=function_title)
 
+    def get_function(
+        self,
+        function_title: str,
+        provider_name: Optional[str],
+    ) -> Optional[Function]:
+        """
+        This method returns the specified function unconditionally.
+
+        Args:
+            function_title (str): title of the function
+            provider_name (str | None): name of the provider owner of the function
+
+        Returns:
+            Program | None: returns the function if it exists
+        """
+
+        queryset = Function.objects.filter(title=function_title)
+
+        if provider_name:
+            queryset = queryset.filter(provider__name=provider_name)
+
+        result_queryset = queryset.first()
+
+        if result_queryset is None:
+            complete_name = (
+                f"{provider_name}/{function_title}" if provider_name else function_title
+            )
+            logger.warning(
+                "Function [%s] was not found",
+                complete_name,
+            )
+
+        return queryset
+
     def get_trial_instances(self, function: Function) -> List[Group]:
         """
         Returns the details of the function groups from trial_instances field

gateway/api/repositories/jobs.py

@@ -2,14 +2,45 @@
 Repository implementation for Job model
 """
 import logging
-from typing import List, Optional
-from django.db.models import Q
+from dataclasses import dataclass
+from datetime import datetime
+from typing import List, Optional, Tuple
+
+from django.db.models import Q, QuerySet
+from django.contrib.auth.models import AbstractUser
+
 from api.models import Job
 from api.models import Program as Function
+from api.views.enums.type_filter import TypeFilter
 
 logger = logging.getLogger("gateway")
 
 
+@dataclass(slots=True)
+class JobFilters:
+    """
+    Filters for Job queries.
+
+    Attributes:
+        function: Function title
+        provider: Provider who owns the function
+        limit: Number of results to return per page
+        offset: Number of results to skip
+        type: Type of job to filter
+        status: Current job status
+        created_after: Filter jobs created after this date
+    """
+
+    function: Optional[str] = None
+    provider: Optional[str] = None
+
+    limit: Optional[TypeFilter] = None
+    offset: Optional[TypeFilter] = None
+    filter: Optional[TypeFilter] = None
+    status: Optional[str] = None
+    created_after: Optional[datetime] = None
+
+
 class JobsRepository:
     """
     The main objective of this class is to manage the access to the Job model
@@ -29,7 +60,7 @@ def get_job_by_id(self, job_id: str) -> Job:
         result_queryset = Job.objects.filter(id=job_id).first()
 
         if result_queryset is None:
-            logger.info("Job [%s] was not found", id)
+            logger.info("Job [%s] was not found", job_id)
 
         return result_queryset
 
@@ -47,54 +78,6 @@ def get_program_jobs(self, function: Function, ordering="-created") -> List[Job]
         function_criteria = Q(program=function)
         return Job.objects.filter(function_criteria).order_by(ordering)
 
-    def get_user_jobs(self, user, ordering="-created") -> List[Job]:
-        """
-        Retrieves jobs created by a specific user.
-
-        Args:
-            user (User): The user whose jobs are to be retrieved.
-            ordering (str, optional): The field to order the results by. Defaults to "-created".
-
-        Returns:
-            List[Jobs]: a list of Jobs
-        """
-        user_criteria = Q(author=user)
-        return Job.objects.filter(user_criteria).order_by(ordering)
-
-    def get_user_jobs_with_provider(self, user, ordering="-created") -> List[Job]:
-        """
-        Retrieves jobs created by a specific user that have an associated provider.
-
-        Args:
-            user (User): The user whose jobs are to be retrieved.
-            ordering (str, optional): The field to order the results by. Defaults to "-created".
-
-        Returns:
-            List[Jobs]: a list of Jobs
-        """
-        user_criteria = Q(author=user)
-        provider_exists_criteria = ~Q(program__provider=None)
-        return Job.objects.filter(user_criteria & provider_exists_criteria).order_by(
-            ordering
-        )
-
-    def get_user_jobs_without_provider(self, user, ordering="-created") -> List[Job]:
-        """
-        Retrieves jobs created by a specific user that do not have an associated provider.
-
-        Args:
-            user (User): The user whose jobs are to be retrieved.
-            ordering (str, optional): The field to order the results by. Defaults to "-created".
-
-        Returns:
-            List[Job]: A queryset of Job objects without a provider.
-        """
-        user_criteria = Q(author=user)
-        provider_not_exists_criteria = Q(program__provider=None)
-        return Job.objects.filter(
-            user_criteria & provider_not_exists_criteria
-        ).order_by(ordering)
-
     def update_job_sub_status(self, job: Job, sub_status: Optional[str]) -> bool:
         """
         Updates the sub-status of a running job.
@@ -128,3 +111,72 @@ def update_job_sub_status(self, job: Job, sub_status: Optional[str]) -> bool:
             )
 
         return updated == 1
+
+    def get_user_jobs(
+        self,
+        user: Optional[AbstractUser] = None,
+        filters: JobFilters = None,
+        ordering: str = "-created",
+    ) -> Tuple[QuerySet[Job], int]:
+        """
+        Get user jobs with optional filters and pagination.
+
+        Args:
+            filters: Optional filters to apply
+            limit: Max number of results
+            offset: Number of results to skip
+            ordering: Field to order by. Default: -created (newest first)
+
+        Returns:
+            (queryset, total_count): Filtered results and total count
+        """
+        queryset = Job.objects.order_by(ordering)
+
+        if filters:
+            queryset = self._apply_filters(queryset, filters=filters, author=user)
+
+        return self._paginate_queryset(queryset, filters.limit, filters.offset)
+
+    def _apply_filters(
+        self,
+        queryset: QuerySet,
+        filters: JobFilters,
+        author: Optional[AbstractUser] = None,
+    ) -> QuerySet:
+        """Apply filters to job queryset."""
+        if author:
+            queryset = queryset.filter(author=author)
+
+        match filters.filter:
+            case TypeFilter.CATALOG:
+                if filters.provider:
+                    queryset = queryset.filter(program__provider=filters.provider)
+                else:
+                    queryset = queryset.exclude(program__provider=None)
+            case TypeFilter.SERVERLESS:
+                queryset = queryset.filter(program__provider=None)
+
+        if filters.status:
+            queryset = queryset.filter(status=filters.status)
+
+        if filters.created_after:
+            queryset = queryset.filter(created__gte=filters.created_after)
+
+        if filters.function:
+            queryset = queryset.filter(program__title=filters.function)
+
+        return queryset
+
+    def _paginate_queryset(
+        self, queryset: QuerySet, limit: int | None, offset: int | None
+    ) -> tuple[QuerySet, int]:
+        """Apply pagination to job queryset."""
+        total_count = queryset.count()
+
+        start = offset or 0
+        end = start + limit if limit else None
+
+        if start >= total_count > 0:
+            return queryset.none(), total_count
+
+        return queryset[start:end], total_count

gateway/api/use_cases/files/provider_list.py

@@ -31,7 +31,7 @@ def execute(self, user: AbstractUser, provider_name: str, function_title: str):
         if provider is None or not ProviderAccessPolicy.can_access(
             user=user, provider=provider
         ):
-            raise NotFoundError(f"Provider {provider_name} doesn't exist.")
+            raise NotFoundError(f"Provider {provider} doesn't exist.")
 
         function = self.function_repository.get_function_by_permission(
             user=user,

gateway/api/use_cases/jobs/list.py

@@ -0,0 +1,43 @@
+"""This module contains the usecase get_jos"""
+import logging
+from typing import List
+
+from django.contrib.auth.models import AbstractUser
+
+from api.domain.exceptions.not_found_error import NotFoundError
+from api.models import Job
+from api.repositories.functions import FunctionRepository
+from api.repositories.jobs import JobsRepository, JobFilters
+
+logger = logging.getLogger("gateway.use_cases.jobs")
+
+
+class JobsListUseCase:
+    """Use case for retrieving user jobs with optional filtering and pagination."""
+
+    function_repository = FunctionRepository()
+    jobs_repository = JobsRepository()
+
+    def execute(self, user: AbstractUser, filters: JobFilters) -> tuple[List[Job], int]:
+        """
+        Retrieve user jobs with optional filters and pagination.
+
+        Returns:
+            tuple[list[Job], int]: (jobs, total_count)
+        """
+        if filters.function:
+            function = self.function_repository.get_function(
+                function_title=filters.function,
+                provider_name=filters.provider,
+            )
+
+            if not function:
+                if filters.provider:
+                    error_message = f"Qiskit Function {filters.provider}/{filters.function} doesn't exist."  # pylint: disable=line-too-long
+                else:
+                    error_message = f"Qiskit Function {filters.function} doesn't exist."
+                raise NotFoundError(error_message)
+
+        queryset, total = self.jobs_repository.get_user_jobs(user=user, filters=filters)
+
+        return list(queryset), total