add catalog/by_title entry-point

client/qiskit_serverless/core/client.py

@@ -167,7 +167,7 @@ class BaseClient(JsonSerializable):
         >>> )
     """
 
-    def __init__(
+    def __init__(  # pylint:  disable=too-many-positional-arguments
         self,
         name: str,
         host: Optional[str] = None,
@@ -386,7 +386,7 @@ class ServerlessClient(BaseClient):
         >>> )
     """
 
-    def __init__(
+    def __init__(  # pylint:  disable=too-many-positional-arguments
         self,
         name: Optional[str] = None,
         host: Optional[str] = None,

client/qiskit_serverless/utils/runtime_service_client.py

@@ -93,7 +93,7 @@ class ServerlessRuntimeService(QiskitRuntimeService):
         QiskitRuntimeService (QiskitRuntimeService): Qiskit runtime service object.
     """
 
-    def run(
+    def run(  # pylint:  disable=too-many-positional-arguments
         self,
         program_id: str,
         inputs: Dict,
@@ -105,7 +105,8 @@ def run(
         session_id: Optional[str] = None,
         start_session: Optional[bool] = False,
     ) -> Union[RuntimeJob, RuntimeJobV2]:
-        runtime_job = super().run(
+        """Run a serverless Runtime service job."""
+        runtime_job = super()._run(
             program_id,
             inputs,
             options,

gateway/api/v1/views.py

@@ -141,3 +141,13 @@ def list(self, request):
     )
     def retrieve(self, request, pk=None):
         return super().retrieve(request, pk)
+
+    @swagger_auto_schema(
+        operation_description="Get a specific public function in the catalog by title",
+        responses={
+            status.HTTP_200_OK: v1_serializers.RetrieveCatalogSerializer(many=False)
+        },
+    )
+    @action(methods=["GET"], detail=False)
+    def get_by_title(self, request):
+        return super().get_by_title(request)

gateway/api/views.py

@@ -854,10 +854,24 @@ def get_retrieve_queryset(self, pk):
 
         if public_group is None:
             logger.error("Public group [%s] does not exist.", self.PUBLIC_GROUP_NAME)
-            return []
+            return None
 
         return Program.objects.filter(id=pk, instances=public_group).first()
 
+    def get_by_title_queryset(self, title, provider_name):
+        """
+        QuerySet to retrieve a specifc public programs in the catalog
+        """
+        public_group = Group.objects.filter(name=self.PUBLIC_GROUP_NAME).first()
+
+        if public_group is None:
+            logger.error("Public group [%s] does not exist.", self.PUBLIC_GROUP_NAME)
+            return None
+
+        return Program.objects.filter(
+            title=title, provider__name=provider_name, instances=public_group
+        ).first()
+
     def list(self, request):
         """List public programs in the catalog:"""
         tracer = trace.get_tracer("gateway.tracer")
@@ -890,3 +904,32 @@ def retrieve(self, request, pk=None):
                 instance, context={"user": user}
             )
         return Response(serializer.data)
+
+    @action(methods=["GET"], detail=False)
+    def get_by_title(self, request):
+        """Get a specific program in the catalog:"""
+        tracer = trace.get_tracer("gateway.tracer")
+        ctx = TraceContextTextMapPropagator().extract(carrier=request.headers)
+        with tracer.start_as_current_span("gateway.catalog.retrieve", context=ctx):
+            title = self.request.query_params.get("title")
+            provider_name = self.request.query_params.get("provider")
+            if not title or not provider:
+                return Response(
+                    {"message": "Missing title or provider name."},
+                    status=status.HTTP_400_BAD_REQUEST,
+                )
+
+            instance = self.get_by_title_queryset(title, provider_name)
+            if instance is None:
+                return Response(
+                    {"message": "Qiskit Function not found."},
+                    status=status.HTTP_404_NOT_FOUND,
+                )
+
+            user = None
+            if request.user and request.user.is_authenticated:
+                user = request.user
+            serializer = self.get_serializer_retrieve_catalog(
+                instance, context={"user": user}
+            )
+        return Response(serializer.data)

gateway/tests/api/test_v1_catalog.py

@@ -158,3 +158,83 @@ def test_catalog_retrieve_with_auth_user_with_run_permission(self):
         self.assertTrue(isinstance(public_function.get("additional_info"), dict))
         self.assertEqual(provider.get("name"), "default")
         self.assertEqual(provider.get("readable_name"), "Default")
+
+    def test_catalog_by_title_non_auth_user(self):
+        """Tests catalog get by title non-authenticated."""
+        url = "/api/v1/catalog/get_by_title/"
+        response = self.client.get(
+            url, {"title": "Public-Function", "provider": "default"}, format="json"
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        public_function = response.data
+        provider = public_function.get("provider")
+        self.assertEqual(public_function.get("available"), False)
+        self.assertEqual(public_function.get("title"), "Public-Function")
+        self.assertEqual(public_function.get("readable_title"), "Public Function")
+        self.assertEqual(public_function.get("type"), Program.APPLICATION)
+        self.assertTrue(isinstance(public_function.get("additional_info"), dict))
+        self.assertEqual(provider.get("name"), "default")
+        self.assertEqual(provider.get("readable_name"), "Default")
+
+    def test_catalog_404_by_title_non_auth_user(self):
+        """Tests catalog get by title a non-existent function as non-authenticated."""
+
+        url = "/api/v1/catalog/get_by_title/"
+        response = self.client.get(
+            url, {"title": "Missing-Function", "provider": "default"}, format="json"
+        )
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_catalog_404_by_title_auth_user(self):
+        """Tests catalog get by title a non-existent function as authenticated."""
+        user = models.User.objects.get(username="test_user")
+        self.client.force_authenticate(user=user)
+
+        url = "/api/v1/catalog/get_by_title/"
+        response = self.client.get(
+            url, {"title": "Missing-Function", "provider": "default"}, format="json"
+        )
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_catalog_by_title_with_auth_user_without_run_permission(self):
+        """Tests catalog get by title as authenticated without run permission."""
+        user = models.User.objects.get(username="test_user")
+        self.client.force_authenticate(user=user)
+
+        url = "/api/v1/catalog/get_by_title/"
+        response = self.client.get(
+            url, {"title": "Public-Function", "provider": "default"}, format="json"
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        public_function = response.data
+        provider = public_function.get("provider")
+        self.assertEqual(public_function.get("available"), False)
+        self.assertEqual(public_function.get("title"), "Public-Function")
+        self.assertEqual(public_function.get("readable_title"), "Public Function")
+        self.assertEqual(public_function.get("type"), Program.APPLICATION)
+        self.assertTrue(isinstance(public_function.get("additional_info"), dict))
+        self.assertEqual(provider.get("name"), "default")
+        self.assertEqual(provider.get("readable_name"), "Default")
+
+    def test_catalog_by_title_with_auth_user_with_run_permission(self):
+        """Tests catalog get by title as authenticated with run permission."""
+        user = models.User.objects.get(username="test_user_2")
+        self.client.force_authenticate(user=user)
+
+        url = "/api/v1/catalog/get_by_title/"
+        response = self.client.get(
+            url, {"title": "Public-Function", "provider": "default"}, format="json"
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        public_function = response.data
+        provider = public_function.get("provider")
+        self.assertEqual(public_function.get("available"), True)
+        self.assertEqual(public_function.get("title"), "Public-Function")
+        self.assertEqual(public_function.get("readable_title"), "Public Function")
+        self.assertEqual(public_function.get("type"), Program.APPLICATION)
+        self.assertTrue(isinstance(public_function.get("additional_info"), dict))
+        self.assertEqual(provider.get("name"), "default")
+        self.assertEqual(provider.get("readable_name"), "Default")