feat(client): bootstrap

Author: vmscw

requirements.txt

@@ -1,5 +1,5 @@
 randomname==0.2.1
 httpx==0.27.0
-dataclasses-json==0.6.4
-dataclasses==0.6
-pytimeparse==1.1.8
+pytimeparse==1.1.8
+attrs = ">=22.2.0"
+python-dateutil = "^2.8.0"

scaleway_qaas_client/.gitignore

@@ -0,0 +1,23 @@
+__pycache__/
+build/
+dist/
+*.egg-info/
+.pytest_cache/
+
+# pyenv
+.python-version
+
+# Environments
+.env
+.venv
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# JetBrains
+.idea/
+
+/coverage.xml
+/.coverage

scaleway_qaas_client/__init__.py

@@ -0,0 +1,34 @@
+from quantum_as_a_service_api_client.models import (
+    ScalewayQaasV1Alpha1Platform as QaaSPlatform,
+)
+from quantum_as_a_service_api_client.models import (
+    ScalewayQaasV1Alpha1PlatformAvailability as QaaSPlatformAvailability,
+)
+from quantum_as_a_service_api_client.models import (
+    ScalewayQaasV1Alpha1PlatformTechnology as QaaSPlatformTechnology,
+)
+from quantum_as_a_service_api_client.models import ScalewayQaasV1Alpha1Job as QaaSJob
+from quantum_as_a_service_api_client.models import (
+    ScalewayQaasV1Alpha1JobResult as QaaSJobResut,
+)
+from quantum_as_a_service_api_client.models import (
+    ScalewayQaasV1Alpha1JobStatus as QaaSJobStatus,
+)
+from quantum_as_a_service_api_client.models import (
+    ScalewayQaasV1Alpha1Session as QaaSSession,
+)
+from quantum_as_a_service_api_client.models import (
+    ScalewayQaasV1Alpha1SessionStatus as QaaSSessionStatus,
+)
+from quantum_as_a_service_api_client.models import (
+    ScalewayQaasV1Alpha1Application as QaaSApplication,
+)
+from quantum_as_a_service_api_client.models import (
+    ScalewayQaasV1Alpha1Process as QaaSProcess,
+)
+from quantum_as_a_service_api_client.models import (
+    ScalewayQaasV1Alpha1ProcessResult as QaaSProcessResult,
+)
+from quantum_as_a_service_api_client.models import (
+    ScalewayQaasV1Alpha1ProcessStatus as QaaSProcessStatus,
+)

scaleway_qaas_client/client.py

@@ -0,0 +1,122 @@
+import json
+
+from typing import List, Optional, Dict
+
+from quantum_as_a_service_api_client.models import (
+    CreateJobBody,
+    CreateJobBodyCircuit,
+    CreateSessionBody,
+    TerminateSessionBody,
+    ScalewayQaasV1Alpha1Platform,
+    ScalewayQaasV1Alpha1Job,
+    ScalewayQaasV1Alpha1JobResult,
+    ScalewayQaasV1Alpha1Session,
+)
+
+from quantum_as_a_service_api_client.api.sessions.create_session import (
+    sync as create_session_sync,
+)
+from quantum_as_a_service_api_client.api.sessions.terminate_session import (
+    sync as terminate_session_sync,
+)
+from quantum_as_a_service_api_client.api.sessions.delete_session import (
+    sync_detailed as delete_session_sync,
+)
+from quantum_as_a_service_api_client.api.platforms.list_platforms import (
+    sync as list_platforms_sync,
+)
+from quantum_as_a_service_api_client.api.jobs.create_job import (
+    sync as create_job_sync,
+)
+from quantum_as_a_service_api_client.api.jobs.get_job import (
+    sync as get_job_sync,
+)
+from quantum_as_a_service_api_client.api.jobs.list_job_results import (
+    sync as list_job_result_sync,
+)
+
+from quantum_as_a_service_api_client.client import Client
+
+
+__DEFAULT_URL = "https://api.scaleway.com/qaas/v1alpha1"
+
+
+class QaaSClient:
+    def __init__(self, project_id: str, token: str, url: str = __DEFAULT_URL):
+        self.__token = token
+        self.__project_id = project_id
+
+        self.__client = Client(
+            headers={"X-Auth-Token": self.__token},
+            base_url=self.__url,
+            timeout=10.0,
+            verify_ssl="https" in url,
+        )
+
+    def list_platforms(self, name: Optional[str]) -> List[ScalewayQaasV1Alpha1Platform]:
+        response = list_platforms_sync(client=self.__client, name=name)
+
+        assert response
+
+        return response.platforms
+
+    def create_session(
+        self,
+        name: str,
+        platform_id: str,
+        deduplication_id: str,
+        max_duration: str,
+        max_idle_duration: str,
+    ) -> ScalewayQaasV1Alpha1Session:
+        session = create_session_sync(
+            client=self.__client,
+            body=CreateSessionBody(
+                project_id=self.__project_id,
+                name=name,
+                platform_id=platform_id,
+                deduplication_id=deduplication_id,
+                max_duration=max_duration,
+                max_idle_duration=max_idle_duration,
+            ),
+        )
+
+        return session
+
+    def terminate_session(self, session_id: str) -> ScalewayQaasV1Alpha1Session:
+        session = terminate_session_sync(
+            client=self.__client,
+            body=TerminateSessionBody(
+                session_id=session_id,
+            ),
+        )
+
+        return session
+
+    def delete_session(self, session_id: str):
+        delete_session_sync(client=self.__client, session_id=session_id)
+
+    def create_job(
+        self, name: str, session_id: str, circuits: Dict
+    ) -> ScalewayQaasV1Alpha1Job:
+        circuits = circuits if isinstance(circuits, str) else json.dumps(circuits)
+
+        job = create_job_sync(
+            client=self.__client,
+            body=CreateJobBody(
+                name=name,
+                session_id=session_id,
+                circuit=CreateJobBodyCircuit(qiskit_circuit=circuits),
+            ),
+        )
+
+        return job
+
+    def get_job(self, job_id: str) -> ScalewayQaasV1Alpha1Job:
+        job = get_job_sync(client=self.__client, job_id=job_id)
+
+        return job
+
+    def list_job_results(self, job_id: str) -> List[ScalewayQaasV1Alpha1JobResult]:
+        response = list_job_result_sync(client=self.__client, job_id=job_id)
+
+        return response.job_results

scaleway_qaas_client/quantum_as_a_service_api_client/__init__.py

@@ -0,0 +1,8 @@
+"""A client library for accessing Quantum as a Service API"""
+
+from .client import AuthenticatedClient, Client
+
+__all__ = (
+    "AuthenticatedClient",
+    "Client",
+)

scaleway_qaas_client/quantum_as_a_service_api_client/api/__init__.py

@@ -0,0 +1 @@
+"""Contains methods for accessing the API"""

scaleway_qaas_client/quantum_as_a_service_api_client/api/applications/__init__.py

@@ -0,0 +1 @@
+"""Contains endpoint functions for accessing the API"""

scaleway_qaas_client/quantum_as_a_service_api_client/api/applications/get_application.py

@@ -0,0 +1,164 @@
+from http import HTTPStatus
+from typing import Any, Optional, Union
+
+import httpx
+
+from ... import errors
+from ...client import AuthenticatedClient, Client
+from ...models.scaleway_qaas_v1_alpha_1_application import (
+    ScalewayQaasV1Alpha1Application,
+)
+from ...types import Response
+
+
+def _get_kwargs(
+    application_id: str,
+) -> dict[str, Any]:
+    _kwargs: dict[str, Any] = {
+        "method": "get",
+        "url": f"/qaas/v1alpha1/applications/{application_id}",
+    }
+
+    return _kwargs
+
+
+def _parse_response(
+    *, client: Union[AuthenticatedClient, Client], response: httpx.Response
+) -> Optional[ScalewayQaasV1Alpha1Application]:
+    if response.status_code == 200:
+        response_200 = ScalewayQaasV1Alpha1Application.from_dict(response.json())
+
+        return response_200
+    if client.raise_on_unexpected_status:
+        raise errors.UnexpectedStatus(response.status_code, response.content)
+    else:
+        return None
+
+
+def _build_response(
+    *, client: Union[AuthenticatedClient, Client], response: httpx.Response
+) -> Response[ScalewayQaasV1Alpha1Application]:
+    return Response(
+        status_code=HTTPStatus(response.status_code),
+        content=response.content,
+        headers=response.headers,
+        parsed=_parse_response(client=client, response=response),
+    )
+
+
+def sync_detailed(
+    application_id: str,
+    *,
+    client: AuthenticatedClient,
+) -> Response[ScalewayQaasV1Alpha1Application]:
+    """Get application information
+
+     Retrieve information about the provided **applcation ID**, such as name, type and compatible
+    platforms.
+
+    Args:
+        application_id (str): Unique ID of the application.
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        Response[ScalewayQaasV1Alpha1Application]
+    """
+
+    kwargs = _get_kwargs(
+        application_id=application_id,
+    )
+
+    response = client.get_httpx_client().request(
+        **kwargs,
+    )
+
+    return _build_response(client=client, response=response)
+
+
+def sync(
+    application_id: str,
+    *,
+    client: AuthenticatedClient,
+) -> Optional[ScalewayQaasV1Alpha1Application]:
+    """Get application information
+
+     Retrieve information about the provided **applcation ID**, such as name, type and compatible
+    platforms.
+
+    Args:
+        application_id (str): Unique ID of the application.
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        ScalewayQaasV1Alpha1Application
+    """
+
+    return sync_detailed(
+        application_id=application_id,
+        client=client,
+    ).parsed
+
+
+async def asyncio_detailed(
+    application_id: str,
+    *,
+    client: AuthenticatedClient,
+) -> Response[ScalewayQaasV1Alpha1Application]:
+    """Get application information
+
+     Retrieve information about the provided **applcation ID**, such as name, type and compatible
+    platforms.
+
+    Args:
+        application_id (str): Unique ID of the application.
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        Response[ScalewayQaasV1Alpha1Application]
+    """
+
+    kwargs = _get_kwargs(
+        application_id=application_id,
+    )
+
+    response = await client.get_async_httpx_client().request(**kwargs)
+
+    return _build_response(client=client, response=response)
+
+
+async def asyncio(
+    application_id: str,
+    *,
+    client: AuthenticatedClient,
+) -> Optional[ScalewayQaasV1Alpha1Application]:
+    """Get application information
+
+     Retrieve information about the provided **applcation ID**, such as name, type and compatible
+    platforms.
+
+    Args:
+        application_id (str): Unique ID of the application.
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        ScalewayQaasV1Alpha1Application
+    """
+
+    return (
+        await asyncio_detailed(
+            application_id=application_id,
+            client=client,
+        )
+    ).parsed