# 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.importbase64importioimporttypingfrompathlibimportPathfromtypingimportAny,Optional,UnionimportplatformdirsimportpydanticaspdtfrompydanticimportConfigDict,GetCoreSchemaHandlerfrompydantic_coreimportCoreSchema,core_schemafromqiskitimportqpyfromqiskit.circuitimportQuantumCircuitfromtyping_extensionsimportSelffromqiskit_aqt_provider.api_clientimportResourcefromqiskit_aqt_provider.aqt_optionsimportAQTOptionsfromqiskit_aqt_provider.utilsimportmap_exceptionsfromqiskit_aqt_provider.versionsimportQISKIT_AQT_PROVIDER_VERSION
[docs]classJobNotFoundError(Exception):"""A job was not found in persistent storage."""
classCircuits:"""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@classmethoddef__get_pydantic_core_schema__(cls,source_type:Any,handler:GetCoreSchemaHandler)->CoreSchema:"""Setup custom validator, to turn this class into a pydantic model."""returncore_schema.no_info_plain_validator_function(function=cls.validate)@classmethoddefvalidate(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. """ifisinstance(value,Circuits):# self bypassreturntyping.cast(Self,value)ifnotisinstance(value,str):raiseValueError(f"Expected string, received {type(value)}")data=base64.b64decode(value.encode("ascii"))buf=io.BytesIO(data)obj=qpy.load(buf)ifnotisinstance(obj,list):obj=[obj]forn,qcinenumerate(obj):ifnotisinstance(qc,QuantumCircuit):raiseValueError(f"Object at position {n} is not a QuantumCircuit: {type(qc)}")returncls(circuits=obj)@classmethoddefjson_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)returnbase64.b64encode(buf.getvalue()).decode("ascii")classJob(pdt.BaseModel):"""Model for job persistence in local storage."""model_config=ConfigDict(frozen=True,json_encoders={Circuits:Circuits.json_encoder})resource:Resourcecircuits:Circuitsoptions:AQTOptions@classmethod@map_exceptions(JobNotFoundError,source_exc=(FileNotFoundError,))defrestore(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")returncls.model_validate_json(data)defpersist(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")returnfilepath@classmethoddefremove_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)@classmethoddeffilepath(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. """returnstore_path/job_iddefget_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. """ifoverrideisnotNone:override.mkdir(parents=True,exist_ok=True)returnoverridereturnPath(platformdirs.user_cache_dir("qiskit_aqt_provider",version=QISKIT_AQT_PROVIDER_VERSION,ensure_exists=True,))