Skip to content

Results

lume_services.results.generic

Attributes

Classes

Result

Bases: BaseModel

Creates a data model for a result and generates a unique result hash.

Attributes
model_type class-attribute 
model_type: str = Field('generic', alias='collection')
id class-attribute 
id: Optional[str] = Field(alias='_id', exclude=True)
flow_id class-attribute 
flow_id: str
inputs class-attribute 
inputs: dict
outputs class-attribute 
outputs: dict
date_modified class-attribute 
date_modified: datetime = datetime.utcnow()
unique_on class-attribute 
unique_on: List[str] = Field(
    ["inputs", "outputs", "flow_id"],
    alias="index",
    exclude=True,
)
unique_hash class-attribute 
unique_hash: str
result_type_string class-attribute 
result_type_string: str
Classes
Config
Attributes
allow_arbitrary_types class-attribute 
allow_arbitrary_types = True
json_encoders class-attribute 
json_encoders = JSON_ENCODERS
allow_population_by_field_name class-attribute 
allow_population_by_field_name = True
extra class-attribute 
extra = Extra.forbid
Functions
validate_all 
validate_all(values)
Source code in lume_services/results/generic.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@root_validator(pre=True)
def validate_all(cls, values):
    unique_fields = cls.__fields__["unique_on"].default

    # If flow_id is not passed, check prefect context
    if not values.get("flow_id"):
        if not context.flow_id:
            raise ValueError("No flow_id passed to result")

        values["flow_id"] = context.flow_id

    # create index hash
    if not values.get("unique_hash"):

        for field in unique_fields:
            if not values.get(field):
                raise ValueError("%s not provided.", field)

        values["unique_hash"] = fingerprint_dict(
            {index: values[index] for index in unique_fields}
        )

    if values.get("_id"):
        id = values["_id"]
        if isinstance(id, (ObjectId,)):
            values["_id"] = str(values["_id"])

    values["result_type_string"] = f"{cls.__module__}:{cls.__name__}"

    return values
get_unique_result_index 
get_unique_result_index() -> dict
Source code in lume_services/results/generic.py 
78
79
def get_unique_result_index(self) -> dict:
    return {field: getattr(self, field) for field in self.unique_on}
insert 
insert(
    results_db_service: ResultsDB = Provide[
        Context.results_db_service
    ],
)
Source code in lume_services/results/generic.py 
81
82
83
84
85
86
87
88
@inject
def insert(
    self, results_db_service: ResultsDB = Provide[Context.results_db_service]
):

    # must convert to jsonable dict
    rep = self.jsonable_dict()
    return results_db_service.insert_one(rep)
load_from_query classmethod 
load_from_query(
    query,
    results_db_service: ResultsDB = Provide[
        Context.results_db_service
    ],
)
Source code in lume_services/results/generic.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
@inject
def load_from_query(
    cls,
    query,
    results_db_service: ResultsDB = Provide[Context.results_db_service],
):
    res = results_db_service.find(
        collection=cls.__fields__["model_type"].default, query=query
    )

    if len(res) == 0:
        raise ValueError("Provided query returned no results. %s", query)

    elif len(res) > 1:
        raise ValueError("Provided query returned multiple results. %s", query)

    return cls(**res[0])
jsonable_dict 
jsonable_dict() -> dict
Source code in lume_services/results/generic.py 
109
110
def jsonable_dict(self) -> dict:
    return json.loads(self.json(by_alias=True))
unique_rep 
unique_rep() -> dict

Get minimal representation needed to load result object from database.

Source code in lume_services/results/generic.py 
112
113
114
115
116
117
def unique_rep(self) -> dict:
    """Get minimal representation needed to load result object from database."""
    return {
        "result_type_string": self.result_type_string,
        "query": self.get_unique_result_index(),
    }

Functions

lume_services.results.impact

Attributes

Classes

ImpactResult

Bases: Result

Extends Result base and implements Impact specific attributes

Attributes
model_type class-attribute 
model_type: str = Field('Impact', alias='collection')
plot_file class-attribute 
plot_file: Optional[ImageFile]
archive class-attribute 
archive: HDF5File
pv_collection_isotime class-attribute 
pv_collection_isotime: datetime
config class-attribute 
config: dict