from __future__ import annotations
from abc import ABC, abstractmethod
from contextlib import contextmanager
from pathlib import Path
from typing import ContextManager, Union
from pharmpy.model import Model
from ...workflows.model_entry import ModelEntry
from ..hashing import ModelHash
from ..results import Results
class ModelTransaction(ABC):
def __init__(self, database: ModelDatabase, obj: Union[Model, ModelEntry, ModelHash]):
self.database = database
if isinstance(obj, ModelEntry):
self.model_entry = obj
self.key = ModelHash(obj.model)
elif isinstance(obj, Model):
self.model_entry = ModelEntry.create(obj)
self.key = ModelHash(obj)
elif isinstance(obj, ModelHash):
self.model_entry = None
self.key = obj
else:
raise ValueError(
f'Invalid type `model_or_model_entry`: got {type(obj)}, expected Model or ModelEntry'
)
@abstractmethod
def store_model(self) -> None:
"""Store the model object bound to this transaction"""
pass
@abstractmethod
def store_local_file(self, path: Path, new_filename: Union[str, None] = None):
"""Store a file from the local machine for the model bound to this
transaction
Parameters
----------
path : Path
Path to file
new_filename: str|None
Filename to give to the file. Optional, defaults to original
filename given by path.
"""
pass
@abstractmethod
def store_metadata(self, metadata) -> None:
"""Store metadata for the model bound to this transaction
Parameters
----------
metadata : Dict
A dictionary with metadata
"""
pass
@abstractmethod
def store_modelfit_results(self) -> None:
"""Store modelfit results of the model bound to this transaction"""
pass
@abstractmethod
def store_model_entry(self) -> None:
"""Store model entry of the model entry bound to this transaction"""
pass
class ModelSnapshot(ABC):
def __init__(self, database: ModelDatabase, model: Union[Model, ModelHash]):
self.database = database
self.key = ModelHash(model)
@abstractmethod
def list_all_files(self) -> list[str]:
"""Lists all file names related to a model run bound to this snapshot
Note that this will not return a list of full paths, only the name of the files
Returns
-------
list[str]
List of file names
"""
pass
@abstractmethod
def retrieve_file(self, filename: str, destination_path: Path, force: bool = False) -> None:
"""Retrieve one file related to a model run bound to this snapshot
Parameters
----------
filename : str
Name of file
destination_path : Path
Destination path
force : bool
Force overwrite of file (default False)
"""
pass
@abstractmethod
def retrieve_all_files(self, destination_path: Path, force: bool = False) -> None:
"""Retrieve all files related to a model run bound to this snapshot
Parameters
----------
destination_path : Path
Destination path
force : bool
Force overwrite of files (default False)
"""
pass
@abstractmethod
def retrieve_model(self) -> Model:
"""Get the model bound to this snapshot
Returns
-------
Model
Retrieved model object
"""
pass
@abstractmethod
def retrieve_modelfit_results(self) -> Results:
"""Read modelfit results from the database
Parameters
----------
name : str
Name of the model
Returns
-------
Results
Retrieved model results object
"""
pass
@abstractmethod
def retrieve_model_entry(self) -> ModelEntry:
"""Read model entry from the database
Returns
-------
ModelEntry
Retrieved model entry object
"""
pass
[docs]
class ModelDatabase(ABC):
"""Baseclass for databases for models and results of model runs
Primary key is the model name, which means that models with
the same name will be counted as the same model.
Currently the model database is centered around storing and retrieving
files belonging to the model or results from fitting the model. This
doesn't mean that implementations have to use a file system to store the
files.
"""
[docs]
@abstractmethod
def store_model(self, model: Model) -> None:
"""Store a model object
Parameters
----------
model : Model
Pharmpy model object
"""
pass
[docs]
@abstractmethod
def store_local_file(
self, model: Union[Model, ModelHash], path: Path, new_filename: Union[str, None] = None
) -> None:
"""Store a file from the local machine
Parameters
----------
model : Model
Pharmpy model object
path : Path
Path to file
new_filename: str|None
Filename to give to the file. Optional, defaults to original
filename given by path.
"""
pass
[docs]
@abstractmethod
def store_modelfit_results(self, model: Union[Model, ModelHash]) -> None:
"""Store modelfit results
Parameters
----------
model : Model
Pharmpy model object
"""
pass
[docs]
@abstractmethod
def store_model_entry(self, model_entry: ModelEntry) -> None:
"""Store model entry of the model entry bound to this transaction
Parameters
----------
model_entry : ModelEntry
Pharmpy ModelEntry object
"""
pass
[docs]
@abstractmethod
def list_all_files(self, model: Union[Model, ModelHash]) -> list[str]:
"""Lists all file names related to a model run
Note that this will not return a list of full paths, only the name of the files
Parameters
----------
model : Model, ModelHash
Model object or ModelHash
Returns
-------
list[str]
List of file names
"""
pass
[docs]
@abstractmethod
def retrieve_file(
self,
model: Union[Model, ModelHash],
filename: str,
destination_path: Path,
force: bool = False,
) -> None:
"""Retrieve one file related to a model run
Parameters
----------
model : Model, ModelHash
Model object or ModelHash
filename : str
Name of file
destination_path : Path
Destination path
force : bool
Force overwrite of file (default False)
"""
pass
[docs]
@abstractmethod
def retrieve_all_files(
self, model: Union[Model, ModelHash], destination_path: Path, force: bool = False
) -> None:
"""Retrieve all files related to a model run
Parameters
----------
model : Model, ModelHash
Model object or ModelHash
destination_path : Path
Destination path
force : bool
Force overwrite of files (default False)
"""
pass
[docs]
@abstractmethod
def retrieve_model(self, model: ModelHash) -> Model:
"""Read a model from the database
Parameters
----------
model_name : str
Name of the model
Returns
-------
Model
Retrieved model object
"""
pass
[docs]
@abstractmethod
def retrieve_modelfit_results(self, model: Union[Model, ModelHash]) -> Results:
"""Read modelfit results from the database
Parameters
----------
model_name : str
Name of the model
Returns
-------
Results
Retrieved model results object
"""
pass
[docs]
@abstractmethod
def retrieve_model_entry(self, model: Union[Model, ModelHash]) -> ModelEntry:
"""Read model entry from the database
Parameters
----------
model_name : str
Name of the model
Returns
-------
ModelEntry
Retrieved model entry object
"""
pass
[docs]
@abstractmethod
def snapshot(self, model: Union[Model, ModelHash]) -> ContextManager[ModelSnapshot]:
"""Creates a readable snapshot context for a given model.
Parameters
----------
model_name : str
Name of the Pharmpy model object
"""
pass
[docs]
@abstractmethod
def transaction(
self, obj: Union[Model, ModelEntry, ModelHash]
) -> ContextManager[ModelTransaction]:
"""Creates a writable transaction context for a given model.
Parameters
----------
obj : Model | ModelEntry | ModelHash
Pharmpy model, ModelEntry or ModelHash object
"""
pass
class NonTransactionalModelDatabase(ModelDatabase):
@contextmanager
def snapshot(self, model_name: str):
yield DummySnapshot(self, model_name)
@contextmanager
def transaction(self, model_or_model_entry: Union[Model, ModelEntry]):
yield DummyTransaction(self, model_or_model_entry)
class DummyTransaction(ModelTransaction):
def __init__(self, database: ModelDatabase, model_or_model_entry: Union[Model, ModelEntry]):
self.db = database
if isinstance(model_or_model_entry, ModelEntry):
self.model_entry = model_or_model_entry
elif isinstance(model_or_model_entry, Model):
self.model_entry = ModelEntry.create(model_or_model_entry)
else:
raise ValueError(
f'Invalid type `model_or_model_entry`: got {type(model_or_model_entry)}, expected Model or ModelEntry'
)
def store_model(self) -> None:
return self.db.store_model(self.model_entry.model)
def store_local_file(self, path: Path, new_filename: Union[str, None] = None) -> None:
return self.db.store_local_file(self.model_entry.model, path, new_filename)
def store_metadata(self, metadata: dict) -> None:
return self.db.store_metadata(self.model_entry.model, metadata)
def store_modelfit_results(self) -> None:
return self.db.store_modelfit_results(self.model_entry.model)
def store_model_entry(self) -> None:
return self.db.store_model_entry(self.model_entry)
class DummySnapshot(ModelSnapshot):
def __init__(self, database: ModelDatabase, model_name: str):
self.db = database
self.name = model_name
def list_all_files(self) -> list[str]:
return self.db.list_all_files(self.name)
def retrieve_file(self, filename: str, destination_path: Path, force: bool = False) -> None:
self.db.retrieve_file(self.name, filename, destination_path, force)
def retrieve_all_files(self, destination_path: Path, force: bool = False) -> None:
self.db.retrieve_all_files(self.name, destination_path, force)
def retrieve_model(self) -> Model:
return self.db.retrieve_model(self.name)
def retrieve_modelfit_results(self) -> Results:
return self.db.retrieve_modelfit_results(self.name)
def retrieve_model_entry(self) -> ModelEntry:
return self.db.retrieve_model_entry(self.name)
class TransactionalModelDatabase(ModelDatabase):
def store_model(self, model: Union[Model, ModelEntry]) -> None:
with self.transaction(model) as txn:
return txn.store_model()
def store_local_file(
self,
model: Union[Model, ModelEntry, ModelHash],
path: Path,
new_filename: Union[str, None] = None,
) -> None:
with self.transaction(model) as txn:
return txn.store_local_file(path, new_filename)
def store_metadata(self, model: Union[Model, ModelEntry, ModelHash], metadata: dict) -> None:
with self.transaction(model) as txn:
return txn.store_metadata(metadata)
def store_modelfit_results(self, model: Union[Model, ModelEntry]) -> None:
with self.transaction(model) as txn:
return txn.store_modelfit_results()
def store_model_entry(self, model_entry: ModelEntry) -> None:
with self.transaction(model_entry) as txn:
return txn.store_model_entry()
def list_all_files(self, obj: Union[Model, ModelEntry, ModelHash]) -> list[str]:
with self.snapshot(obj) as sn:
return sn.list_all_files()
def retrieve_file(
self,
obj: Union[Model, ModelEntry, ModelHash],
filename: str,
destination_path: Path,
force: bool = False,
) -> None:
with self.snapshot(obj) as sn:
sn.retrieve_file(filename, destination_path, force)
def retrieve_all_files(
self, obj: Union[Model, ModelEntry, ModelHash], destination_path, force: bool = False
) -> None:
with self.snapshot(obj) as sn:
sn.retrieve_all_files(destination_path, force)
def retrieve_model(self, obj: Union[Model, ModelEntry, ModelHash]) -> Model:
with self.snapshot(obj) as sn:
return sn.retrieve_model()
def retrieve_modelfit_results(self, obj: Union[Model, ModelEntry, ModelHash]) -> Results:
with self.snapshot(obj) as sn:
return sn.retrieve_modelfit_results()
def retrieve_model_entry(self, obj: Union[Model, ModelEntry, ModelHash]) -> ModelEntry:
with self.snapshot(obj) as sn:
return sn.retrieve_model_entry()
class PendingTransactionError(Exception):
pass
