9255177207

Committed 27 May 2024 12:52PM UTC coverage: 99.736%. Remained the same

Build # 9255177207

Build Type

push

github

Committed by

web-flow

Commit Message

Prepare release 1.5.0 (#161)

Run Details

2267 of 2273 relevant lines covered (99.74%)

3.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.77

/qiskit_aqt_provider/persistence.py

# This code is part of Qiskit.
#
# (C) Copyright Alpine Quantum Technologies 2023
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.

import base64
import io
import typing
from pathlib import Path
from typing import Any, Optional, Union

import platformdirs
import pydantic as pdt
from pydantic import ConfigDict, GetCoreSchemaHandler
from pydantic_core import CoreSchema, core_schema
from qiskit import qpy
from qiskit.circuit import QuantumCircuit
from typing_extensions import Self

from qiskit_aqt_provider.api_models import ResourceId
from qiskit_aqt_provider.aqt_options import AQTOptions
from qiskit_aqt_provider.utils import map_exceptions
from qiskit_aqt_provider.versions import QISKIT_AQT_PROVIDER_VERSION


class JobNotFoundError(Exception):
    """A job was not found in persistent storage."""


class Circuits:
    """Custom Pydantic type to persist and restore lists of Qiskit circuits.

    Serialization of :class:`QuantumCircuit <qiskit.circuit.QuantumCircuit>` instances is
    provided by :mod:`qiskit.qpy`.
    """

    def __init__(self, circuits: list[QuantumCircuit]) -> None:
        """Initialize a container filled with the given circuits."""
        self.circuits = circuits

    @classmethod
    def __get_pydantic_core_schema__(
        cls, source_type: Any, handler: GetCoreSchemaHandler
    ) -> CoreSchema:
        """Setup custom validator, to turn this class into a pydantic model."""
        return core_schema.no_info_plain_validator_function(function=cls.validate)

    @classmethod
    def validate(cls, value: Union[Self, str]) -> Self:
        """Parse the base64-encoded :mod:`qiskit.qpy` representation of a list of quantum circuits.

        Because initializing a Pydantic model also triggers validation, this parser accepts
        already formed instances of this class and returns them unvalidated.
        """
        if isinstance(value, Circuits):  # self bypass
            return typing.cast(Self, value)

        if not isinstance(value, str):
            raise ValueError(f"Expected string, received {type(value)}")

        data = base64.b64decode(value.encode("ascii"))
        buf = io.BytesIO(data)
        obj = qpy.load(buf)

        if not isinstance(obj, list):
            obj = [obj]

        for n, qc in enumerate(obj):
            if not isinstance(qc, QuantumCircuit):
                raise ValueError(f"Object at position {n} is not a QuantumCircuit: {type(qc)}")

        return cls(circuits=obj)

    @classmethod
    def json_encoder(cls, value: Self) -> str:
        """Return a base64-encoded QPY representation of the held list of circuits."""
        buf = io.BytesIO()
        qpy.dump(value.circuits, buf)
        return base64.b64encode(buf.getvalue()).decode("ascii")


class Job(pdt.BaseModel):
    """Model for job persistence in local storage."""

    model_config = ConfigDict(frozen=True, json_encoders={Circuits: Circuits.json_encoder})

    resource: ResourceId
    circuits: Circuits
    options: AQTOptions

    @classmethod
    @map_exceptions(JobNotFoundError, source_exc=(FileNotFoundError,))
    def restore(cls, job_id: str, store_path: Path) -> Self:
        """Load data for a job by ID from local storage.

        Args:
            job_id: identifier of the job to restore.
            store_path: path to the local storage directory.

        Raises:
            JobNotFoundError: no job with the given identifier is stored in the local storage.
        """
        data = cls.filepath(job_id, store_path).read_text("utf-8")
        return cls.model_validate_json(data)

    def persist(self, job_id: str, store_path: Path) -> Path:
        """Persist the job data to the local storage.

        Args:
            job_id: storage key for this job data.
            store_path: path to the local storage directory.

        Returns:
            The path of the persisted data file.
        """
        filepath = self.filepath(job_id, store_path)
        filepath.write_text(self.model_dump_json(), "utf-8")
        return filepath

    @classmethod
    def remove_from_store(cls, job_id: str, store_path: Path) -> None:
        """Remove persisted job data from the local storage.

        This function also succeeds if there is no data under `job_id`.

        Args:
            job_id: storage key for the data to delete.
            store_path: path to the local storage directory.
        """
        cls.filepath(job_id, store_path).unlink(missing_ok=True)

    @classmethod
    def filepath(cls, job_id: str, store_path: Path) -> Path:
        """Path of the file to store data under a given key in local storage.

        Args:
            job_id: storage key for the data.
            store_path: path to the local storage directory.
        """
        return store_path / job_id


def get_store_path(override: Optional[Path] = None) -> Path:
    """Resolve the local persistence store path.

    By default, this is the user cache directory for this package.
    Different cache directories are used for different package versions.

    Args:
        override: if given, return this override instead of the default path.

    Returns:
       Path for the persistence store. Ensured to exist.
    """
    if override is not None:
        override.mkdir(parents=True, exist_ok=True)
        return override

    return Path(
        platformdirs.user_cache_dir(
            "qiskit_aqt_provider",
            version=QISKIT_AQT_PROVIDER_VERSION,
            ensure_exists=True,
        )
    )

1	# This code is part of Qiskit.
2	#
3	# (C) Copyright Alpine Quantum Technologies 2023
4	#
5	# This code is licensed under the Apache License, Version 2.0. You may
6	# obtain a copy of this license in the LICENSE.txt file in the root directory
7	# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
8	#
9	# Any modifications or derivative works of this code must retain this
10	# copyright notice, and modified files need to carry a notice indicating
11	# that they have been altered from the originals.
12
13	import base64	4✔
14	import io	4✔
15	import typing	4✔
16	from pathlib import Path	4✔
17	from typing import Any, Optional, Union	4✔
18
19	import platformdirs	4✔
20	import pydantic as pdt	4✔
21	from pydantic import ConfigDict, GetCoreSchemaHandler	4✔
22	from pydantic_core import CoreSchema, core_schema	4✔
23	from qiskit import qpy	4✔
24	from qiskit.circuit import QuantumCircuit	4✔
25	from typing_extensions import Self	4✔
26
27	from qiskit_aqt_provider.api_models import ResourceId	4✔
28	from qiskit_aqt_provider.aqt_options import AQTOptions	4✔
29	from qiskit_aqt_provider.utils import map_exceptions	4✔
30	from qiskit_aqt_provider.versions import QISKIT_AQT_PROVIDER_VERSION	4✔
31
32
33	class JobNotFoundError(Exception):	4✔
34	"""A job was not found in persistent storage."""	4✔
35
36
37	class Circuits:	4✔
38	"""Custom Pydantic type to persist and restore lists of Qiskit circuits.	4✔
39
40	Serialization of :class:`QuantumCircuit <qiskit.circuit.QuantumCircuit>` instances is
41	provided by :mod:`qiskit.qpy`.
42	"""
43
44	def __init__(self, circuits: list[QuantumCircuit]) -> None:	4✔
45	"""Initialize a container filled with the given circuits."""
46	self.circuits = circuits	4✔
47
48	@classmethod	4✔
49	def __get_pydantic_core_schema__(	4✔
50	cls, source_type: Any, handler: GetCoreSchemaHandler
51	) -> CoreSchema:
52	"""Setup custom validator, to turn this class into a pydantic model."""
53	return core_schema.no_info_plain_validator_function(function=cls.validate)	4✔
54
55	@classmethod	4✔
56	def validate(cls, value: Union[Self, str]) -> Self:	4✔
57	"""Parse the base64-encoded :mod:`qiskit.qpy` representation of a list of quantum circuits.
58
59	Because initializing a Pydantic model also triggers validation, this parser accepts
60	already formed instances of this class and returns them unvalidated.
61	"""
62	if isinstance(value, Circuits): # self bypass	4✔
63	return typing.cast(Self, value)	4✔
64
65	if not isinstance(value, str):	4✔
66	raise ValueError(f"Expected string, received {type(value)}")	×
67
68	data = base64.b64decode(value.encode("ascii"))	4✔
69	buf = io.BytesIO(data)	4✔
70	obj = qpy.load(buf)	4✔
71
72	if not isinstance(obj, list):	4✔
73	obj = [obj]	×
74
75	for n, qc in enumerate(obj):	4✔
76	if not isinstance(qc, QuantumCircuit):	4✔
77	raise ValueError(f"Object at position {n} is not a QuantumCircuit: {type(qc)}")	×
78
79	return cls(circuits=obj)	4✔
80
81	@classmethod	4✔
82	def json_encoder(cls, value: Self) -> str:	4✔
83	"""Return a base64-encoded QPY representation of the held list of circuits."""
84	buf = io.BytesIO()	4✔
85	qpy.dump(value.circuits, buf)	4✔
86	return base64.b64encode(buf.getvalue()).decode("ascii")	4✔
87
88
89	class Job(pdt.BaseModel):	4✔
90	"""Model for job persistence in local storage."""	4✔
91
92	model_config = ConfigDict(frozen=True, json_encoders={Circuits: Circuits.json_encoder})	4✔
93
94	resource: ResourceId	4✔
95	circuits: Circuits	4✔
96	options: AQTOptions	4✔
97
98	@classmethod	4✔
99	@map_exceptions(JobNotFoundError, source_exc=(FileNotFoundError,))	4✔
100	def restore(cls, job_id: str, store_path: Path) -> Self:	4✔
101	"""Load data for a job by ID from local storage.
102
103	Args:
104	job_id: identifier of the job to restore.
105	store_path: path to the local storage directory.
106
107	Raises:
108	JobNotFoundError: no job with the given identifier is stored in the local storage.
109	"""
110	data = cls.filepath(job_id, store_path).read_text("utf-8")	4✔
111	return cls.model_validate_json(data)	4✔
112
113	def persist(self, job_id: str, store_path: Path) -> Path:	4✔
114	"""Persist the job data to the local storage.
115
116	Args:
117	job_id: storage key for this job data.
118	store_path: path to the local storage directory.
119
120	Returns:
121	The path of the persisted data file.
122	"""
123	filepath = self.filepath(job_id, store_path)	4✔
124	filepath.write_text(self.model_dump_json(), "utf-8")	4✔
125	return filepath	4✔
126
127	@classmethod	4✔
128	def remove_from_store(cls, job_id: str, store_path: Path) -> None:	4✔
129	"""Remove persisted job data from the local storage.
130
131	This function also succeeds if there is no data under `job_id`.
132
133	Args:
134	job_id: storage key for the data to delete.
135	store_path: path to the local storage directory.
136	"""
137	cls.filepath(job_id, store_path).unlink(missing_ok=True)	4✔
138
139	@classmethod	4✔
140	def filepath(cls, job_id: str, store_path: Path) -> Path:	4✔
141	"""Path of the file to store data under a given key in local storage.
142
143	Args:
144	job_id: storage key for the data.
145	store_path: path to the local storage directory.
146	"""
147	return store_path / job_id	4✔
148
149
150	def get_store_path(override: Optional[Path] = None) -> Path:	4✔
151	"""Resolve the local persistence store path.
152
153	By default, this is the user cache directory for this package.
154	Different cache directories are used for different package versions.
155
156	Args:
157	override: if given, return this override instead of the default path.
158
159	Returns:
160	Path for the persistence store. Ensured to exist.
161	"""
162	if override is not None:	4✔
163	override.mkdir(parents=True, exist_ok=True)	4✔
164	return override	4✔
165
166	return Path(	4✔
167	platformdirs.user_cache_dir(
168	"qiskit_aqt_provider",
169	version=QISKIT_AQT_PROVIDER_VERSION,
170	ensure_exists=True,
171	)
172	)

qiskit-community / qiskit-aqt-provider / 9255177207

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous