/
generic_assets.py
130 lines (107 loc) · 4.23 KB
/
generic_assets.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
from typing import Optional
import json
from marshmallow import validates, validates_schema, ValidationError, fields
from flexmeasures.data import ma
from flexmeasures.data.models.user import Account
from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType
from flexmeasures.data.schemas.utils import (
FMValidationError,
MarshmallowClickMixin,
with_appcontext_if_needed,
)
class JSON(fields.Field):
def _deserialize(self, value, attr, data, **kwargs):
if value:
try:
return json.loads(value)
except ValueError:
return None
return None
def _serialize(self, value, attr, data, **kwargs):
return json.dumps(value)
class GenericAssetSchema(ma.SQLAlchemySchema):
"""
GenericAsset schema, with validations.
"""
id = ma.auto_field(dump_only=True)
name = fields.Str(required=True)
account_id = ma.auto_field()
latitude = ma.auto_field()
longitude = ma.auto_field()
generic_asset_type_id = fields.Integer(required=True)
attributes = JSON(required=False)
class Meta:
model = GenericAsset
@validates_schema(skip_on_field_errors=False)
def validate_name_is_unique_in_account(self, data, **kwargs):
if "name" in data and "account_id" in data:
asset = GenericAsset.query.filter(
GenericAsset.name == data["name"]
and GenericAsset.account_id == data["account_id"]
).one_or_none()
if asset:
raise ValidationError(
f"An asset with the name {data['name']} already exists in this account.",
"name",
)
@validates("generic_asset_type_id")
def validate_generic_asset_type(self, generic_asset_type_id: int):
generic_asset_type = GenericAssetType.query.get(generic_asset_type_id)
if not generic_asset_type:
raise ValidationError(
f"GenericAssetType with id {generic_asset_type_id} doesn't exist."
)
@validates("account_id")
def validate_account(self, account_id: int):
account = Account.query.get(account_id)
if not account:
raise ValidationError(f"Account with Id {account_id} doesn't exist.")
@validates("latitude")
def validate_latitude(self, latitude: Optional[float]):
"""Validate optional latitude."""
if latitude is None:
return
if latitude < -90:
raise ValidationError(
f"Latitude {latitude} exceeds the minimum latitude of -90 degrees."
)
if latitude > 90:
raise ValidationError(
f"Latitude {latitude} exceeds the maximum latitude of 90 degrees."
)
@validates("longitude")
def validate_longitude(self, longitude: Optional[float]):
"""Validate optional longitude."""
if longitude is None:
return
if longitude < -180:
raise ValidationError(
f"Longitude {longitude} exceeds the minimum longitude of -180 degrees."
)
if longitude > 180:
raise ValidationError(
f"Longitude {longitude} exceeds the maximum longitude of 180 degrees."
)
class GenericAssetTypeSchema(ma.SQLAlchemySchema):
"""
GenericAssetType schema, with validations.
"""
id = ma.auto_field()
name = fields.Str()
description = ma.auto_field()
class Meta:
model = GenericAssetType
class GenericAssetIdField(MarshmallowClickMixin, fields.Int):
"""Field that deserializes to a GenericAsset and serializes back to an integer."""
@with_appcontext_if_needed()
def _deserialize(self, value, attr, obj, **kwargs) -> GenericAsset:
"""Turn a generic asset id into a GenericAsset."""
generic_asset = GenericAsset.query.get(value)
if generic_asset is None:
raise FMValidationError(f"No asset found with id {value}.")
# lazy loading now (asset is somehow not in session after this)
generic_asset.generic_asset_type
return generic_asset
def _serialize(self, asset, attr, data, **kwargs):
"""Turn a GenericAsset into a generic asset id."""
return asset.id