7567300363

Committed 18 Jan 2024 08:32AM UTC coverage: 99.83%. Remained the same

Build # 7567300363

Build Type

push

github

Committed by

web-flow

Commit Message

Prepare for release 1.1.0 (#120)

Run Details

2353 of 2357 relevant lines covered (99.83%)

3.97 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, List, 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:
        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, List, 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	self.circuits = circuits	4✔
46
47	@classmethod	4✔
48	def __get_pydantic_core_schema__(	4✔
49	cls, source_type: Any, handler: GetCoreSchemaHandler
50	) -> CoreSchema:
51	"""Setup custom validator, to turn this class into a pydantic model."""
52	return core_schema.no_info_plain_validator_function(function=cls.validate)	4✔
53
54	@classmethod	4✔
55	def validate(cls, value: Union[Self, str]) -> Self:	4✔
56	"""Parse the base64-encoded :mod:`qiskit.qpy` representation of a list of quantum circuits.
57
58	Because initializing a Pydantic model also triggers validation, this parser accepts
59	already formed instances of this class and returns them unvalidated.
60	"""
61	if isinstance(value, Circuits): # self bypass	4✔
62	return typing.cast(Self, value)	4✔
63
64	if not isinstance(value, str):	4✔
65	raise ValueError(f"Expected string, received {type(value)}")	×
66
67	data = base64.b64decode(value.encode("ascii"))	4✔
68	buf = io.BytesIO(data)	4✔
69	obj = qpy.load(buf)	4✔
70
71	if not isinstance(obj, list):	4✔
72	obj = [obj]	×
73
74	for n, qc in enumerate(obj):	4✔
75	if not isinstance(qc, QuantumCircuit):	4✔
76	raise ValueError(f"Object at position {n} is not a QuantumCircuit: {type(qc)}")	×
77
78	return cls(circuits=obj)	4✔
79
80	@classmethod	4✔
81	def json_encoder(cls, value: Self) -> str:	4✔
82	"""Return a base64-encoded QPY representation of the held list of circuits."""
83	buf = io.BytesIO()	4✔
84	qpy.dump(value.circuits, buf)	4✔
85	return base64.b64encode(buf.getvalue()).decode("ascii")	4✔
86
87
88	class Job(pdt.BaseModel):	4✔
89	"""Model for job persistence in local storage."""	4✔
90
91	model_config = ConfigDict(frozen=True, json_encoders={Circuits: Circuits.json_encoder})	4✔
92
93	resource: ResourceId	4✔
94	circuits: Circuits	4✔
95	options: AQTOptions	4✔
96
97	@classmethod	4✔
98	@map_exceptions(JobNotFoundError, source_exc=(FileNotFoundError,))	4✔
99	def restore(cls, job_id: str, store_path: Path) -> Self:	4✔
100	"""Load data for a job by ID from local storage.
101
102	Args:
103	job_id: identifier of the job to restore.
104	store_path: path to the local storage directory.
105
106	Raises:
107	JobNotFoundError: no job with the given identifier is stored in the local storage.
108	"""
109	data = cls.filepath(job_id, store_path).read_text("utf-8")	4✔
110	return cls.model_validate_json(data)	4✔
111
112	def persist(self, job_id: str, store_path: Path) -> Path:	4✔
113	"""Persist the job data to the local storage.
114
115	Args:
116	job_id: storage key for this job data.
117	store_path: path to the local storage directory.
118
119	Returns:
120	The path of the persisted data file.
121	"""
122	filepath = self.filepath(job_id, store_path)	4✔
123	filepath.write_text(self.model_dump_json(), "utf-8")	4✔
124	return filepath	4✔
125
126	@classmethod	4✔
127	def remove_from_store(cls, job_id: str, store_path: Path) -> None:	4✔
128	"""Remove persisted job data from the local storage.
129
130	This function also succeeds if there is no data under `job_id`.
131
132	Args:
133	job_id: storage key for the data to delete.
134	store_path: path to the local storage directory.
135	"""
136	cls.filepath(job_id, store_path).unlink(missing_ok=True)	4✔
137
138	@classmethod	4✔
139	def filepath(cls, job_id: str, store_path: Path) -> Path:	4✔
140	"""Path of the file to store data under a given key in local storage.
141
142	Args:
143	job_id: storage key for the data.
144	store_path: path to the local storage directory.
145	"""
146	return store_path / job_id	4✔
147
148
149	def get_store_path(override: Optional[Path] = None) -> Path:	4✔
150	"""Resolve the local persistence store path.
151
152	By default, this is the user cache directory for this package.
153	Different cache directories are used for different package versions.
154
155	Args:
156	override: if given, return this override instead of the default path.
157
158	Returns:
159	Path for the persistence store. Ensured to exist.
160	"""
161	if override is not None:	4✔
162	override.mkdir(parents=True, exist_ok=True)	4✔
163	return override	4✔
164
165	return Path(	4✔
166	platformdirs.user_cache_dir(
167	"qiskit_aqt_provider",
168	version=QISKIT_AQT_PROVIDER_VERSION,
169	ensure_exists=True,
170	)
171	)

qiskit-community / qiskit-aqt-provider / 7567300363

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