9255690342

Committed 27 May 2024 01:27PM UTC coverage: 66.479% (-0.08%) from 66.56%

Build # 9255690342

Build Type

Pull #1072

github

Committed by

nrozanov

Commit Message

Migration file + small fixes

Signed-off-by: Nikolai Rozanov <nickolay.rozanov@gmail.com>

Pull Request Pull Request #1072: Asset audit log

Run Details

5823 of 9634 branches covered (60.44%)

Branch coverage included in aggregate %.

59 of 94 new or added lines in 7 files covered. (62.77%)

71 existing lines in 4 files now uncovered.

7423 of 10291 relevant lines covered (72.13%)

5.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

41.35

/flexmeasures/cli/data_edit.py

"""
CLI commands for editing data
"""

from __future__ import annotations

from datetime import timedelta

import click
import pandas as pd
from flask import current_app as app
from flask.cli import with_appcontext
from flask_security import current_user
import json
from flexmeasures.data.models.user import Account
from flexmeasures.data.schemas.account import AccountIdField
from sqlalchemy import delete

from flexmeasures import Sensor, Asset
from flexmeasures.data import db
from flexmeasures.data.schemas.attributes import validate_special_attributes
from flexmeasures.data.schemas.generic_assets import GenericAssetIdField
from flexmeasures.data.schemas.sensors import SensorIdField
from flexmeasures.data.models.generic_assets import GenericAsset
from flexmeasures.data.models.audit_log import AssetAuditLog
from flexmeasures.data.models.time_series import TimedBelief
from flexmeasures.data.utils import save_to_db
from flexmeasures.cli.utils import MsgStyle, DeprecatedOption, DeprecatedOptionsCommand
from flexmeasures.utils.time_utils import server_now


@click.group("edit")
def fm_edit_data():
    """FlexMeasures: Edit data."""


@fm_edit_data.command("attribute", cls=DeprecatedOptionsCommand)
@with_appcontext
@click.option(
    "--asset",
    "--asset-id",
    "assets",
    required=False,
    multiple=True,
    type=GenericAssetIdField(),
    cls=DeprecatedOption,
    deprecated=["--asset-id"],
    preferred="--asset",
    help="Add/edit attribute to this asset. Follow up with the asset's ID.",
)
@click.option(
    "--sensor",
    "--sensor-id",
    "sensors",
    required=False,
    multiple=True,
    type=SensorIdField(),
    cls=DeprecatedOption,
    deprecated=["--sensor-id"],
    preferred="--sensor",
    help="Add/edit attribute to this sensor. Follow up with the sensor's ID.",
)
@click.option(
    "--attribute",
    "attribute_key",
    required=True,
    help="Add/edit this attribute. Follow up with the name of the attribute.",
)
@click.option(
    "--float",
    "attribute_float_value",
    required=False,
    type=float,
    help="Set the attribute to this float value.",
)
@click.option(
    "--bool",
    "attribute_bool_value",
    required=False,
    type=bool,
    help="Set the attribute to this bool value.",
)
@click.option(
    "--str",
    "attribute_str_value",
    required=False,
    type=str,
    help="Set the attribute to this string value.",
)
@click.option(
    "--int",
    "attribute_int_value",
    required=False,
    type=int,
    help="Set the attribute to this integer value.",
)
@click.option(
    "--list",
    "attribute_list_value",
    required=False,
    type=str,
    help="Set the attribute to this list value. Pass a string with a JSON-parse-able list representation, e.g. '[1,\"a\"]'.",
)
@click.option(
    "--dict",
    "attribute_dict_value",
    required=False,
    type=str,
    help="Set the attribute to this dict value. Pass a string with a JSON-parse-able dict representation, e.g. '{1:\"a\"}'.",
)
@click.option(
    "--null",
    "attribute_null_value",
    required=False,
    is_flag=True,
    default=False,
    help="Set the attribute to a null value.",
)
def edit_attribute(
    attribute_key: str,
    assets: list[GenericAsset],
    sensors: list[Sensor],
    attribute_null_value: bool,
    attribute_float_value: float | None = None,
    attribute_bool_value: bool | None = None,
    attribute_str_value: str | None = None,
    attribute_int_value: int | None = None,
    attribute_list_value: str | None = None,
    attribute_dict_value: str | None = None,
):
    """Edit (or add) an asset attribute or sensor attribute."""

    if not assets and not sensors:
        raise ValueError("Missing flag: pass at least one --asset-id or --sensor-id.")

    # Parse attribute value
    attribute_value = parse_attribute_value(
        attribute_float_value=attribute_float_value,
        attribute_bool_value=attribute_bool_value,
        attribute_str_value=attribute_str_value,
        attribute_int_value=attribute_int_value,
        attribute_list_value=attribute_list_value,
        attribute_dict_value=attribute_dict_value,
        attribute_null_value=attribute_null_value,
    )

    # Some attributes with special in meaning in FlexMeasures must pass validation
    validate_special_attributes(attribute_key, attribute_value)

    # Set attribute
    for asset in assets:
        write_audit_log_for_attribute_update(
            attribute_key, attribute_value, "asset", asset
        )
        asset.attributes[attribute_key] = attribute_value
        db.session.add(asset)
    for sensor in sensors:
        write_audit_log_for_attribute_update(
            attribute_key, attribute_value, "sensor", sensor
        )
        sensor.attributes[attribute_key] = attribute_value
        db.session.add(sensor)
    db.session.commit()
    click.secho("Successfully edited/added attribute.", **MsgStyle.SUCCESS)


@fm_edit_data.command("resample-data", cls=DeprecatedOptionsCommand)
@with_appcontext
@click.option(
    "--sensor",
    "--sensor-id",
    "sensor_ids",
    multiple=True,
    required=True,
    cls=DeprecatedOption,
    deprecated=["--sensor-id"],
    preferred="--sensor",
    help="Resample data for this sensor. Follow up with the sensor's ID. This argument can be given multiple times.",
)
@click.option(
    "--event-resolution",
    "event_resolution_in_minutes",
    type=int,
    required=True,
    help="New event resolution as an integer number of minutes.",
)
@click.option(
    "--from",
    "start_str",
    required=False,
    help="Resample only data from this datetime onwards. Follow up with a timezone-aware datetime in ISO 6801 format.",
)
@click.option(
    "--until",
    "end_str",
    required=False,
    help="Resample only data until this datetime. Follow up with a timezone-aware datetime in ISO 6801 format.",
)
@click.option(
    "--skip-integrity-check",
    is_flag=True,
    help="Whether to skip checking the resampled time series data for each sensor."
    " By default, an excerpt and the mean value of the original"
    " and resampled data will be shown for manual approval.",
)
def resample_sensor_data(
    sensor_ids: list[int],
    event_resolution_in_minutes: int,
    start_str: str | None = None,
    end_str: str | None = None,
    skip_integrity_check: bool = False,
):
    """Assign a new event resolution to an existing sensor and resample its data accordingly."""
    event_resolution = timedelta(minutes=event_resolution_in_minutes)
    event_starts_after = pd.Timestamp(start_str)  # note that "" or None becomes NaT
    event_ends_before = pd.Timestamp(end_str)
    current_user_id, current_user_name = None, None
    if current_user.is_authenticated:
        current_user_id, current_user_name = current_user.id, current_user.username
    for sensor_id in sensor_ids:
        sensor = db.session.get(Sensor, sensor_id)
        if sensor.event_resolution == event_resolution:
            click.echo(f"{sensor} already has the desired event resolution.")
            continue
        df_original = sensor.search_beliefs(
            most_recent_beliefs_only=False,
            event_starts_after=event_starts_after,
            event_ends_before=event_ends_before,
        ).sort_values("event_start")
        df_resampled = df_original.resample_events(event_resolution).sort_values(
            "event_start"
        )
        if not skip_integrity_check:
            message = ""
            if sensor.event_resolution < event_resolution:
                message += f"Downsampling {sensor} to {event_resolution} will result in a loss of data. "
            click.confirm(
                message
                + f"Data before:\n{df_original}\nData after:\n{df_resampled}\nMean before: {df_original['event_value'].mean()}\nMean after: {df_resampled['event_value'].mean()}\nContinue?",
                abort=True,
            )

        audit_log = AssetAuditLog(
            event_datetime=server_now(),
            event=f"Resampled sensor data for sensor '{sensor.name}': {sensor.id} to {event_resolution} from {sensor.event_resolution}",
            active_user_id=current_user_id,
            active_user_name=current_user_name,
            affected_asset_id=sensor.generic_asset_id,
        )
        db.session.add(audit_log)

        # Update sensor
        sensor.event_resolution = event_resolution
        db.session.add(sensor)

        # Update sensor data
        query = delete(TimedBelief).filter_by(sensor=sensor)
        if not pd.isnull(event_starts_after):
            query = query.filter(TimedBelief.event_start >= event_starts_after)
        if not pd.isnull(event_ends_before):
            query = query.filter(
                TimedBelief.event_start + sensor.event_resolution <= event_ends_before
            )
        db.session.execute(query)
        save_to_db(df_resampled, bulk_save_objects=True)
    db.session.commit()
    click.secho("Successfully resampled sensor data.", **MsgStyle.SUCCESS)


@fm_edit_data.command("transfer-ownership")
@with_appcontext
@click.option(
    "--asset",
    "asset",
    type=GenericAssetIdField(),
    required=True,
    help="Change the ownership of this asset and its children. Follow up with the asset's ID.",
)
@click.option(
    "--new-owner",
    "new_owner",
    type=AccountIdField(),
    required=True,
    help="New owner of the asset and its children.",
)
def transfer_ownership(asset: Asset, new_owner: Account):
    """
    Transfer the ownership of and asset and its children to an account.
    """

    current_user_id, current_user_name = None, None
    if current_user.is_authenticated:
        current_user_id, current_user_name = current_user.id, current_user.username

    def transfer_ownership_recursive(asset: Asset, account: Account):
        audit_log = AssetAuditLog(
            event_datetime=server_now(),
            event=f"Transfered ownership for asset '{asset.name}': {asset.id} from {asset.owner.id} to {account.id}",
            active_user_id=current_user_id,
            active_user_name=current_user_name,
            affected_asset_id=asset.id,
        )
        db.session.add(audit_log)

        asset.owner = account
        for child in asset.child_assets:
            transfer_ownership_recursive(child, account)

    transfer_ownership_recursive(asset, new_owner)
    click.secho(
        f"Success! Asset `{asset}` ownership was transfered to account `{new_owner}`.",
        **MsgStyle.SUCCESS,
    )

    db.session.commit()


app.cli.add_command(fm_edit_data)


def parse_attribute_value(  # noqa: C901
    attribute_null_value: bool,
    attribute_float_value: float | None = None,
    attribute_bool_value: bool | None = None,
    attribute_str_value: str | None = None,
    attribute_int_value: int | None = None,
    attribute_list_value: str | None = None,
    attribute_dict_value: str | None = None,
) -> float | int | bool | str | list | dict | None:
    """Parse attribute value."""
    if not single_true(
        [attribute_null_value]
        + [
            v is not None
            for v in [
                attribute_float_value,
                attribute_bool_value,
                attribute_str_value,
                attribute_int_value,
                attribute_list_value,
                attribute_dict_value,
            ]
        ]
    ):
        raise ValueError("Cannot set multiple values simultaneously.")
    if attribute_null_value:
        return None
    elif attribute_float_value is not None:
        return float(attribute_float_value)
    elif attribute_bool_value is not None:
        return bool(attribute_bool_value)
    elif attribute_int_value is not None:
        return int(attribute_int_value)
    elif attribute_list_value is not None:
        try:
            val = json.loads(attribute_list_value)
        except json.decoder.JSONDecodeError as jde:
            raise ValueError(f"Error parsing list value: {jde}")
        if not isinstance(val, list):
            raise ValueError(f"{val} is not a list.")
        return val
    elif attribute_dict_value is not None:
        try:
            val = json.loads(attribute_dict_value)
        except json.decoder.JSONDecodeError as jde:
            raise ValueError(f"Error parsing dict value: {jde}")
        if not isinstance(val, dict):
            raise ValueError(f"{val} is not a dict.")
        return val
    return attribute_str_value


def single_true(iterable) -> bool:
    i = iter(iterable)
    return any(i) and not any(i)


def write_audit_log_for_attribute_update(
    attribute_key, attribute_value, entity_type, asset_or_sensor
):
    current_user_id, current_user_name = None, None
    if (
        current_user
        and hasattr(current_user, "is_authenticated")
        and current_user.is_authenticated
    ):
        current_user_id, current_user_name = current_user.id, current_user.username

    old_value = asset_or_sensor.attributes.get(attribute_key)
    if entity_type == "sensor":
        event = f"Updated sensor '{asset_or_sensor.name}': {asset_or_sensor.id} "
        affected_asset_id = (asset_or_sensor.generic_asset_id,)
    else:
        event = f"Updated asset '{asset_or_sensor.name}': {asset_or_sensor.id} "
        affected_asset_id = asset_or_sensor.id
    event += f"attribute '{attribute_key}' to {attribute_value} from {old_value}"

    audit_log = AssetAuditLog(
        event_datetime=server_now(),
        event=event,
        active_user_id=current_user_id,
        active_user_name=current_user_name,
        affected_asset_id=affected_asset_id,
    )
    db.session.add(audit_log)

1	"""	4✔
2	CLI commands for editing data
3	"""
4
5	from __future__ import annotations	8✔
6
7	from datetime import timedelta	8✔
8
9	import click	8✔
10	import pandas as pd	8✔
11	from flask import current_app as app	8✔
12	from flask.cli import with_appcontext	8✔
13	from flask_security import current_user	8✔
14	import json	8✔
15	from flexmeasures.data.models.user import Account	8✔
16	from flexmeasures.data.schemas.account import AccountIdField	8✔
17	from sqlalchemy import delete	8✔
18
19	from flexmeasures import Sensor, Asset	8✔
20	from flexmeasures.data import db	8✔
21	from flexmeasures.data.schemas.attributes import validate_special_attributes	8✔
22	from flexmeasures.data.schemas.generic_assets import GenericAssetIdField	8✔
23	from flexmeasures.data.schemas.sensors import SensorIdField	8✔
24	from flexmeasures.data.models.generic_assets import GenericAsset	8✔
25	from flexmeasures.data.models.audit_log import AssetAuditLog	8✔
26	from flexmeasures.data.models.time_series import TimedBelief	8✔
27	from flexmeasures.data.utils import save_to_db	8✔
28	from flexmeasures.cli.utils import MsgStyle, DeprecatedOption, DeprecatedOptionsCommand	8✔
29	from flexmeasures.utils.time_utils import server_now	8✔
30
31
32	@click.group("edit")	8✔
33	def fm_edit_data():	8✔
34	"""FlexMeasures: Edit data."""
35
36
37	@fm_edit_data.command("attribute", cls=DeprecatedOptionsCommand)	8✔
38	@with_appcontext	8✔
39	@click.option(	8✔
40	"--asset",
41	"--asset-id",
42	"assets",
43	required=False,
44	multiple=True,
45	type=GenericAssetIdField(),
46	cls=DeprecatedOption,
47	deprecated=["--asset-id"],
48	preferred="--asset",
49	help="Add/edit attribute to this asset. Follow up with the asset's ID.",
50	)
51	@click.option(	8✔
52	"--sensor",
53	"--sensor-id",
54	"sensors",
55	required=False,
56	multiple=True,
57	type=SensorIdField(),
58	cls=DeprecatedOption,
59	deprecated=["--sensor-id"],
60	preferred="--sensor",
61	help="Add/edit attribute to this sensor. Follow up with the sensor's ID.",
62	)
63	@click.option(	8✔
64	"--attribute",
65	"attribute_key",
66	required=True,
67	help="Add/edit this attribute. Follow up with the name of the attribute.",
68	)
69	@click.option(	8✔
70	"--float",
71	"attribute_float_value",
72	required=False,
73	type=float,
74	help="Set the attribute to this float value.",
75	)
76	@click.option(	8✔
77	"--bool",
78	"attribute_bool_value",
79	required=False,
80	type=bool,
81	help="Set the attribute to this bool value.",
82	)
83	@click.option(	8✔
84	"--str",
85	"attribute_str_value",
86	required=False,
87	type=str,
88	help="Set the attribute to this string value.",
89	)
90	@click.option(	8✔
91	"--int",
92	"attribute_int_value",
93	required=False,
94	type=int,
95	help="Set the attribute to this integer value.",
96	)
97	@click.option(	8✔
98	"--list",
99	"attribute_list_value",
100	required=False,
101	type=str,
102	help="Set the attribute to this list value. Pass a string with a JSON-parse-able list representation, e.g. '[1,\"a\"]'.",
103	)
104	@click.option(	8✔
105	"--dict",
106	"attribute_dict_value",
107	required=False,
108	type=str,
109	help="Set the attribute to this dict value. Pass a string with a JSON-parse-able dict representation, e.g. '{1:\"a\"}'.",
110	)
111	@click.option(	8✔
112	"--null",
113	"attribute_null_value",
114	required=False,
115	is_flag=True,
116	default=False,
117	help="Set the attribute to a null value.",
118	)
119	def edit_attribute(	8✔
120	attribute_key: str,
121	assets: list[GenericAsset],
122	sensors: list[Sensor],
123	attribute_null_value: bool,
124	attribute_float_value: float \| None = None,
125	attribute_bool_value: bool \| None = None,
126	attribute_str_value: str \| None = None,
127	attribute_int_value: int \| None = None,
128	attribute_list_value: str \| None = None,
129	attribute_dict_value: str \| None = None,
130	):
131	"""Edit (or add) an asset attribute or sensor attribute."""
132
UNCOV 133	if not assets and not sensors:	×
UNCOV 134	raise ValueError("Missing flag: pass at least one --asset-id or --sensor-id.")	×
135
136	# Parse attribute value
UNCOV 137	attribute_value = parse_attribute_value(	×
138	attribute_float_value=attribute_float_value,
139	attribute_bool_value=attribute_bool_value,
140	attribute_str_value=attribute_str_value,
141	attribute_int_value=attribute_int_value,
142	attribute_list_value=attribute_list_value,
143	attribute_dict_value=attribute_dict_value,
144	attribute_null_value=attribute_null_value,
145	)
146
147	# Some attributes with special in meaning in FlexMeasures must pass validation
UNCOV 148	validate_special_attributes(attribute_key, attribute_value)	×
149
150	# Set attribute
151	for asset in assets:	×
NEW 152	write_audit_log_for_attribute_update(	×
153	attribute_key, attribute_value, "asset", asset
154	)
155	asset.attributes[attribute_key] = attribute_value	×
156	db.session.add(asset)	×
157	for sensor in sensors:	×
NEW 158	write_audit_log_for_attribute_update(	×
159	attribute_key, attribute_value, "sensor", sensor
160	)
161	sensor.attributes[attribute_key] = attribute_value	×
162	db.session.add(sensor)	×
UNCOV 163	db.session.commit()	×
UNCOV 164	click.secho("Successfully edited/added attribute.", **MsgStyle.SUCCESS)	×
165
166
167	@fm_edit_data.command("resample-data", cls=DeprecatedOptionsCommand)	8✔
168	@with_appcontext	8✔
169	@click.option(	8✔
170	"--sensor",
171	"--sensor-id",
172	"sensor_ids",
173	multiple=True,
174	required=True,
175	cls=DeprecatedOption,
176	deprecated=["--sensor-id"],
177	preferred="--sensor",
178	help="Resample data for this sensor. Follow up with the sensor's ID. This argument can be given multiple times.",
179	)
180	@click.option(	8✔
181	"--event-resolution",
182	"event_resolution_in_minutes",
183	type=int,
184	required=True,
185	help="New event resolution as an integer number of minutes.",
186	)
187	@click.option(	8✔
188	"--from",
189	"start_str",
190	required=False,
191	help="Resample only data from this datetime onwards. Follow up with a timezone-aware datetime in ISO 6801 format.",
192	)
193	@click.option(	8✔
194	"--until",
195	"end_str",
196	required=False,
197	help="Resample only data until this datetime. Follow up with a timezone-aware datetime in ISO 6801 format.",
198	)
199	@click.option(	8✔
200	"--skip-integrity-check",
201	is_flag=True,
202	help="Whether to skip checking the resampled time series data for each sensor."
203	" By default, an excerpt and the mean value of the original"
204	" and resampled data will be shown for manual approval.",
205	)
206	def resample_sensor_data(	8✔
207	sensor_ids: list[int],
208	event_resolution_in_minutes: int,
209	start_str: str \| None = None,
210	end_str: str \| None = None,
211	skip_integrity_check: bool = False,
212	):
213	"""Assign a new event resolution to an existing sensor and resample its data accordingly."""
214	event_resolution = timedelta(minutes=event_resolution_in_minutes)	×
215	event_starts_after = pd.Timestamp(start_str) # note that "" or None becomes NaT	×
216	event_ends_before = pd.Timestamp(end_str)	×
NEW 217	current_user_id, current_user_name = None, None	×
NEW 218	if current_user.is_authenticated:	×
NEW 219	current_user_id, current_user_name = current_user.id, current_user.username	×
220	for sensor_id in sensor_ids:	×
UNCOV 221	sensor = db.session.get(Sensor, sensor_id)	×
UNCOV 222	if sensor.event_resolution == event_resolution:	×
UNCOV 223	click.echo(f"{sensor} already has the desired event resolution.")	×
UNCOV 224	continue	×
225	df_original = sensor.search_beliefs(	×
226	most_recent_beliefs_only=False,
227	event_starts_after=event_starts_after,
228	event_ends_before=event_ends_before,
229	).sort_values("event_start")
230	df_resampled = df_original.resample_events(event_resolution).sort_values(	×
231	"event_start"
232	)
UNCOV 233	if not skip_integrity_check:	×
UNCOV 234	message = ""	×
UNCOV 235	if sensor.event_resolution < event_resolution:	×
UNCOV 236	message += f"Downsampling {sensor} to {event_resolution} will result in a loss of data. "	×
UNCOV 237	click.confirm(	×
238	message
239	+ f"Data before:\n{df_original}\nData after:\n{df_resampled}\nMean before: {df_original['event_value'].mean()}\nMean after: {df_resampled['event_value'].mean()}\nContinue?",
240	abort=True,
241	)
242
NEW 243	audit_log = AssetAuditLog(	×
244	event_datetime=server_now(),
245	event=f"Resampled sensor data for sensor '{sensor.name}': {sensor.id} to {event_resolution} from {sensor.event_resolution}",
246	active_user_id=current_user_id,
247	active_user_name=current_user_name,
248	affected_asset_id=sensor.generic_asset_id,
249	)
NEW 250	db.session.add(audit_log)	×
251
252	# Update sensor
UNCOV 253	sensor.event_resolution = event_resolution	×
UNCOV 254	db.session.add(sensor)	×
255
256	# Update sensor data
257	query = delete(TimedBelief).filter_by(sensor=sensor)	×
258	if not pd.isnull(event_starts_after):	×
UNCOV 259	query = query.filter(TimedBelief.event_start >= event_starts_after)	×
UNCOV 260	if not pd.isnull(event_ends_before):	×
UNCOV 261	query = query.filter(	×
262	TimedBelief.event_start + sensor.event_resolution <= event_ends_before
263	)
UNCOV 264	db.session.execute(query)	×
UNCOV 265	save_to_db(df_resampled, bulk_save_objects=True)	×
UNCOV 266	db.session.commit()	×
UNCOV 267	click.secho("Successfully resampled sensor data.", **MsgStyle.SUCCESS)	×
268
269
270	@fm_edit_data.command("transfer-ownership")	8✔
271	@with_appcontext	8✔
272	@click.option(	8✔
273	"--asset",
274	"asset",
275	type=GenericAssetIdField(),
276	required=True,
277	help="Change the ownership of this asset and its children. Follow up with the asset's ID.",
278	)
279	@click.option(	8✔
280	"--new-owner",
281	"new_owner",
282	type=AccountIdField(),
283	required=True,
284	help="New owner of the asset and its children.",
285	)
286	def transfer_ownership(asset: Asset, new_owner: Account):	8✔
287	"""
288	Transfer the ownership of and asset and its children to an account.
289	"""
290
291	current_user_id, current_user_name = None, None	8✔
292	if current_user.is_authenticated:	8✔
NEW UNCOV 293	current_user_id, current_user_name = current_user.id, current_user.username	×
294
295	def transfer_ownership_recursive(asset: Asset, account: Account):	8✔
296	audit_log = AssetAuditLog(	8✔
297	event_datetime=server_now(),
298	event=f"Transfered ownership for asset '{asset.name}': {asset.id} from {asset.owner.id} to {account.id}",
299	active_user_id=current_user_id,
300	active_user_name=current_user_name,
301	affected_asset_id=asset.id,
302	)
303	db.session.add(audit_log)	8✔
304
305	asset.owner = account	8✔
306	for child in asset.child_assets:	8✔
307	transfer_ownership_recursive(child, account)	8✔
308
309	transfer_ownership_recursive(asset, new_owner)	8✔
310	click.secho(	8✔
311	f"Success! Asset `{asset}` ownership was transfered to account `{new_owner}`.",
312	**MsgStyle.SUCCESS,
313	)
314
315	db.session.commit()	8✔
316
317
318	app.cli.add_command(fm_edit_data)	8✔
319
320
321	def parse_attribute_value( # noqa: C901	8✔
322	attribute_null_value: bool,
323	attribute_float_value: float \| None = None,
324	attribute_bool_value: bool \| None = None,
325	attribute_str_value: str \| None = None,
326	attribute_int_value: int \| None = None,
327	attribute_list_value: str \| None = None,
328	attribute_dict_value: str \| None = None,
329	) -> float \| int \| bool \| str \| list \| dict \| None:
330	"""Parse attribute value."""
331	if not single_true(	×
332	[attribute_null_value]
333	+ [
334	v is not None
335	for v in [
336	attribute_float_value,
337	attribute_bool_value,
338	attribute_str_value,
339	attribute_int_value,
340	attribute_list_value,
341	attribute_dict_value,
342	]
343	]
344	):
345	raise ValueError("Cannot set multiple values simultaneously.")	×
346	if attribute_null_value:	×
347	return None	×
348	elif attribute_float_value is not None:	×
349	return float(attribute_float_value)	×
350	elif attribute_bool_value is not None:	×
351	return bool(attribute_bool_value)	×
352	elif attribute_int_value is not None:	×
353	return int(attribute_int_value)	×
UNCOV 354	elif attribute_list_value is not None:	×
UNCOV 355	try:	×
UNCOV 356	val = json.loads(attribute_list_value)	×
357	except json.decoder.JSONDecodeError as jde:	×
358	raise ValueError(f"Error parsing list value: {jde}")	×
UNCOV 359	if not isinstance(val, list):	×
UNCOV 360	raise ValueError(f"{val} is not a list.")	×
UNCOV 361	return val	×
UNCOV 362	elif attribute_dict_value is not None:	×
UNCOV 363	try:	×
UNCOV 364	val = json.loads(attribute_dict_value)	×
UNCOV 365	except json.decoder.JSONDecodeError as jde:	×
UNCOV 366	raise ValueError(f"Error parsing dict value: {jde}")	×
UNCOV 367	if not isinstance(val, dict):	×
UNCOV 368	raise ValueError(f"{val} is not a dict.")	×
UNCOV 369	return val	×
UNCOV 370	return attribute_str_value	×
371
372
373	def single_true(iterable) -> bool:	8✔
UNCOV 374	i = iter(iterable)	×
UNCOV 375	return any(i) and not any(i)	×
376
377
378	def write_audit_log_for_attribute_update(	8✔
379	attribute_key, attribute_value, entity_type, asset_or_sensor
380	):
NEW UNCOV 381	current_user_id, current_user_name = None, None	×
NEW UNCOV 382	if (	×
383	current_user
384	and hasattr(current_user, "is_authenticated")
385	and current_user.is_authenticated
386	):
NEW UNCOV 387	current_user_id, current_user_name = current_user.id, current_user.username	×
388
NEW UNCOV 389	old_value = asset_or_sensor.attributes.get(attribute_key)	×
NEW UNCOV 390	if entity_type == "sensor":	×
NEW UNCOV 391	event = f"Updated sensor '{asset_or_sensor.name}': {asset_or_sensor.id} "	×
NEW UNCOV 392	affected_asset_id = (asset_or_sensor.generic_asset_id,)	×
393	else:
NEW UNCOV 394	event = f"Updated asset '{asset_or_sensor.name}': {asset_or_sensor.id} "	×
NEW UNCOV 395	affected_asset_id = asset_or_sensor.id	×
NEW UNCOV 396	event += f"attribute '{attribute_key}' to {attribute_value} from {old_value}"	×
397
NEW UNCOV 398	audit_log = AssetAuditLog(	×
399	event_datetime=server_now(),
400	event=event,
401	active_user_id=current_user_id,
402	active_user_name=current_user_name,
403	affected_asset_id=affected_asset_id,
404	)
NEW UNCOV 405	db.session.add(audit_log)	×

FlexMeasures / flexmeasures / 9255690342

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