7528577372

Committed 15 Jan 2024 12:04PM UTC coverage: 66.625% (+0.04%) from 66.582%

Build # 7528577372

Build Type

Pull #938

github

Committed by

Ahmad-Wahid

Commit Message

first check for a table if exists then data

Signed-off-by: Ahmad Wahid <ahmedwahid16101@gmail.com>

Pull Request Pull Request #938: Support SQLAlchemy 2.0.x

Run Details

3773 of 6348 branches covered (0.0%)

Branch coverage included in aggregate %.

119 of 172 new or added lines in 45 files covered. (69.19%)

9 existing lines in 8 files now uncovered.

6811 of 9538 relevant lines covered (71.41%)

2.14 hits per line

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

69.89

/flexmeasures/ui/crud/assets.py

from __future__ import annotations

import copy
import json

from flask import url_for, current_app
from flask_classful import FlaskView
from flask_wtf import FlaskForm
from flask_security import login_required, current_user
from wtforms import StringField, DecimalField, SelectField
from wtforms.validators import DataRequired, optional
from sqlalchemy import select
from flexmeasures.auth.policy import user_has_admin_access

from flexmeasures.data import db
from flexmeasures.auth.error_handling import unauthorized_handler
from flexmeasures.auth.policy import check_access
from flexmeasures.data.models.generic_assets import (
    GenericAssetType,
    GenericAsset,
    get_center_location_of_assets,
)
from flexmeasures.data.models.user import Account
from flexmeasures.data.models.time_series import Sensor
from flexmeasures.ui.utils.view_utils import render_flexmeasures_template
from flexmeasures.ui.crud.api_wrapper import InternalApi


"""
Asset crud view.

Note: This uses the internal dev API version
      ― if those endpoints get moved or updated to a higher version,
      we probably should change the version used here, as well.
"""


class AssetForm(FlaskForm):
    """The default asset form only allows to edit the name and location."""

    name = StringField("Name")
    latitude = DecimalField(
        "Latitude",
        validators=[optional()],
        places=None,
        render_kw={"placeholder": "--Click the map or enter a latitude--"},
    )
    longitude = DecimalField(
        "Longitude",
        validators=[optional()],
        places=None,
        render_kw={"placeholder": "--Click the map or enter a longitude--"},
    )
    attributes = StringField("Other attributes (JSON)", default="{}")

    def validate_on_submit(self):
        if (
            hasattr(self, "generic_asset_type_id")
            and self.generic_asset_type_id.data == -1
        ):
            self.generic_asset_type_id.data = (
                ""  # cannot be coerced to int so will be flagged as invalid input
            )
        if hasattr(self, "account_id") and self.account_id.data == -1:
            del self.account_id  # asset will be public
        return super().validate_on_submit()

    def to_json(self) -> dict:
        """turn form data into a JSON we can POST to our internal API"""
        data = copy.copy(self.data)
        if data.get("longitude") is not None:
            data["longitude"] = float(data["longitude"])
        if data.get("latitude") is not None:
            data["latitude"] = float(data["latitude"])

        if "csrf_token" in data:
            del data["csrf_token"]

        return data

    def process_api_validation_errors(self, api_response: dict):
        """Process form errors from the API for the WTForm"""
        if not isinstance(api_response, dict):
            return
        for error_header in ("json", "validation_errors"):
            if error_header not in api_response:
                continue
            for field in list(self._fields.keys()):
                if field in list(api_response[error_header].keys()):
                    field_errors = api_response[error_header][field]
                    if isinstance(field_errors, list):
                        self._fields[field].errors += api_response[error_header][field]
                    else:
                        self._fields[field].errors.append(
                            api_response[error_header][field]
                        )


class NewAssetForm(AssetForm):
    """Here, in addition, we allow to set asset type and account."""

    generic_asset_type_id = SelectField(
        "Asset type", coerce=int, validators=[DataRequired()]
    )
    account_id = SelectField("Account", coerce=int)


def with_options(form: AssetForm | NewAssetForm) -> AssetForm | NewAssetForm:
    if "generic_asset_type_id" in form:
        form.generic_asset_type_id.choices = [(-1, "--Select type--")] + [
            (atype.id, atype.name)
            for atype in db.session.scalars(select(GenericAssetType)).all()
        ]
    if "account_id" in form:
        form.account_id.choices = [(-1, "--Select account--")] + [
            (account.id, account.name)
            for account in db.session.scalars(select(Account)).all()
        ]
    return form


def process_internal_api_response(
    asset_data: dict, asset_id: int | None = None, make_obj=False
) -> GenericAsset | dict:
    """
    Turn data from the internal API into something we can use to further populate the UI.
    Either as an asset object or a dict for form filling.

    If we add other data by querying the database, we make sure the asset is not in the session afterwards.
    """

    def expunge_asset():
        # use if no insert is wanted from a previous query which flushes its results
        if asset in db.session:
            db.session.expunge(asset)

    asset_data.pop("status", None)  # might have come from requests.response
    if asset_id:
        asset_data["id"] = asset_id
    if make_obj:
        asset = GenericAsset(
            **{
                **asset_data,
                **{"attributes": json.loads(asset_data.get("attributes", "{}"))},
            }
        )  # TODO: use schema?
        asset.generic_asset_type = db.session.get(
            GenericAssetType, asset.generic_asset_type_id
        )
        expunge_asset()
        asset.owner = db.session.get(Account, asset_data["account_id"])
        expunge_asset()
        if "id" in asset_data:
            asset.sensors = db.session.scalars(
                select(Sensor).filter_by(generic_asset_id=asset_data["id"])
            ).all()
            expunge_asset()
        if asset_data.get("parent_asset_id", None) is not None:
            asset.parent_asset = GenericAsset.query.filter(
                GenericAsset.id == asset_data["parent_asset_id"]
            ).one_or_none()
            expunge_asset()
        return asset
    return asset_data


def user_can_create_assets() -> bool:
    try:
        check_access(current_user.account, "create-children")
    except Exception:
        return False
    return True


def user_can_delete(asset) -> bool:
    try:
        check_access(asset, "delete")
    except Exception:
        return False
    return True


def get_assets_by_account(account_id: int | str | None) -> list[GenericAsset]:
    if account_id is not None:
        get_assets_response = InternalApi().get(
            url_for("AssetAPI:index"), query={"account_id": account_id}
        )
    else:
        get_assets_response = InternalApi().get(url_for("AssetAPI:public"))
    return [
        process_internal_api_response(ad, make_obj=True)
        for ad in get_assets_response.json()
    ]


class AssetCrudUI(FlaskView):
    """
    These views help us offer a Jinja2-based UI.
    The main focus on logic is the API, so these views simply call the API functions,
    and deal with the response.
    Some new functionality, like fetching accounts and asset types, is added here.
    """

    route_base = "/assets"

    @login_required
    def index(self, msg=""):
        """GET from /assets

        List the user's assets. For admins, list across all accounts.
        """
        assets = []

        if user_has_admin_access(current_user, "read"):
            for account in db.session.scalars(select(Account)).all():
                assets += get_assets_by_account(account.id)
            assets += get_assets_by_account(account_id=None)
        else:
            assets = get_assets_by_account(current_user.account_id)

        return render_flexmeasures_template(
            "crud/assets.html",
            assets=assets,
            message=msg,
            user_can_create_assets=user_can_create_assets(),
        )

    @login_required
    def owned_by(self, account_id: str):
        """/assets/owned_by/<account_id>"""
        msg = ""
        get_assets_response = InternalApi().get(
            url_for("AssetAPI:index"),
            query={"account_id": account_id},
            do_not_raise_for=[404],
        )
        if get_assets_response.status_code == 404:
            assets = []
            msg = f"Account {account_id} unknown."
        else:
            assets = [
                process_internal_api_response(ad, make_obj=True)
                for ad in get_assets_response.json()
            ]
        return render_flexmeasures_template(
            "crud/assets.html",
            account=db.session.get(Account, account_id),
            assets=assets,
            msg=msg,
            user_can_create_assets=user_can_create_assets(),
        )

    @login_required
    def get(self, id: str):
        """GET from /assets/<id> where id can be 'new' (and thus the form for asset creation is shown)"""

        if id == "new":
            if not user_can_create_assets():
                return unauthorized_handler(None, [])

            asset_form = with_options(NewAssetForm())
            return render_flexmeasures_template(
                "crud/asset_new.html",
                asset_form=asset_form,
                msg="",
                map_center=get_center_location_of_assets(user=current_user),
                mapboxAccessToken=current_app.config.get("MAPBOX_ACCESS_TOKEN", ""),
            )

        get_asset_response = InternalApi().get(url_for("AssetAPI:fetch_one", id=id))
        asset_dict = get_asset_response.json()

        asset_form = with_options(AssetForm())

        asset = process_internal_api_response(asset_dict, int(id), make_obj=True)
        asset_form.process(data=process_internal_api_response(asset_dict))

        return render_flexmeasures_template(
            "crud/asset.html",
            asset=asset,
            asset_form=asset_form,
            msg="",
            mapboxAccessToken=current_app.config.get("MAPBOX_ACCESS_TOKEN", ""),
            user_can_create_assets=user_can_create_assets(),
            user_can_delete_asset=user_can_delete(asset),
        )

    @login_required
    def post(self, id: str):
        """POST to /assets/<id>, where id can be 'create' (and thus a new asset is made from POST data)
        Most of the code deals with creating a user for the asset if no existing is chosen.
        """

        asset: GenericAsset = None
        error_msg = ""

        if id == "create":
            asset_form = with_options(NewAssetForm())

            account, account_error = _set_account(asset_form)
            asset_type, asset_type_error = _set_asset_type(asset_form)

            form_valid = asset_form.validate_on_submit()

            # Fill up the form with useful errors for the user
            if account_error is not None:
                form_valid = False
                asset_form.account_id.errors.append(account_error)
            if asset_type_error is not None:
                form_valid = False
                asset_form.generic_asset_type_id.errors.append(asset_type_error)

            # Create new asset or return the form for new assets with a message
            if form_valid and asset_type is not None:
                post_asset_response = InternalApi().post(
                    url_for("AssetAPI:post"),
                    args=asset_form.to_json(),
                    do_not_raise_for=[400, 422],
                )
                if post_asset_response.status_code in (200, 201):
                    asset_dict = post_asset_response.json()
                    asset = process_internal_api_response(
                        asset_dict, int(asset_dict["id"]), make_obj=True
                    )
                    msg = "Creation was successful."
                else:
                    current_app.logger.error(
                        f"Internal asset API call unsuccessful [{post_asset_response.status_code}]: {post_asset_response.text}"
                    )
                    asset_form.process_api_validation_errors(post_asset_response.json())
                    if "message" in post_asset_response.json():
                        asset_form.process_api_validation_errors(
                            post_asset_response.json()["message"]
                        )
                        if "json" in post_asset_response.json()["message"]:
                            error_msg = str(
                                post_asset_response.json()["message"]["json"]
                            )
            if asset is None:
                msg = "Cannot create asset. " + error_msg
                return render_flexmeasures_template(
                    "crud/asset_new.html",
                    asset_form=asset_form,
                    msg=msg,
                    map_center=get_center_location_of_assets(user=current_user),
                    mapboxAccessToken=current_app.config.get("MAPBOX_ACCESS_TOKEN", ""),
                )

        else:
            asset_form = with_options(AssetForm())
            if not asset_form.validate_on_submit():
                asset = db.session.get(GenericAsset, id)
                # Display the form data, but set some extra data which the page wants to show.
                asset_info = asset_form.to_json()
                asset_info["id"] = id
                asset_info["account_id"] = asset.account_id
                asset = process_internal_api_response(
                    asset_info, int(id), make_obj=True
                )

                return render_flexmeasures_template(
                    "crud/asset.html",
                    asset_form=asset_form,
                    asset=asset,
                    msg="Cannot edit asset.",
                    mapboxAccessToken=current_app.config.get("MAPBOX_ACCESS_TOKEN", ""),
                    user_can_create_assets=user_can_create_assets(),
                    user_can_delete_asset=user_can_delete(asset),
                )
            patch_asset_response = InternalApi().patch(
                url_for("AssetAPI:patch", id=id),
                args=asset_form.to_json(),
                do_not_raise_for=[400, 422],
            )
            asset_dict = patch_asset_response.json()
            if patch_asset_response.status_code in (200, 201):
                asset = process_internal_api_response(
                    asset_dict, int(id), make_obj=True
                )
                msg = "Editing was successful."
            else:
                current_app.logger.error(
                    f"Internal asset API call unsuccessful [{patch_asset_response.status_code}]: {patch_asset_response.text}"
                )
                msg = "Cannot edit asset."
                asset_form.process_api_validation_errors(
                    patch_asset_response.json().get("message")
                )
                asset = db.session.get(GenericAsset, id)

        return render_flexmeasures_template(
            "crud/asset.html",
            asset=asset,
            asset_form=asset_form,
            msg=msg,
            mapboxAccessToken=current_app.config.get("MAPBOX_ACCESS_TOKEN", ""),
            user_can_create_assets=user_can_create_assets(),
            user_can_delete_asset=user_can_delete(asset),
        )

    @login_required
    def delete_with_data(self, id: str):
        """Delete via /assets/delete_with_data/<id>"""
        InternalApi().delete(url_for("AssetAPI:delete", id=id))
        return self.index(
            msg=f"Asset {id} and assorted meter readings / forecasts have been deleted."
        )


def _set_account(asset_form: NewAssetForm) -> tuple[Account | None, str | None]:
    """Set an account for the to-be-created asset.
    Return the account (if available) and an error message"""
    account_error = None

    if asset_form.account_id.data == -1:
        if user_has_admin_access(current_user, "update"):
            return None, None  # Account can be None (public asset)
        else:
            account_error = "Please pick an existing account."

    account = db.session.execute(
        select(Account).filter_by(id=int(asset_form.account_id.data))
    ).scalar_one_or_none()

    if account:
        asset_form.account_id.data = account.id
    else:
        current_app.logger.error(account_error)
    return account, account_error


def _set_asset_type(
    asset_form: NewAssetForm,
) -> tuple[GenericAssetType | None, str | None]:
    """Set an asset type for the to-be-created asset.
    Return the asset type (if available) and an error message."""
    asset_type = None
    asset_type_error = None

    if int(asset_form.generic_asset_type_id.data) == -1:
        asset_type_error = "Pick an existing asset type."
    else:
        asset_type = db.session.execute(
            select(GenericAssetType).filter_by(
                id=int(asset_form.generic_asset_type_id.data)
            )
        ).scalar_one_or_none()

    if asset_type:
        asset_form.generic_asset_type_id.data = asset_type.id
    else:
        current_app.logger.error(asset_type_error)
    return asset_type, asset_type_error

1	from __future__ import annotations	3✔
2
3	import copy	3✔
4	import json	3✔
5
6	from flask import url_for, current_app	3✔
7	from flask_classful import FlaskView	3✔
8	from flask_wtf import FlaskForm	3✔
9	from flask_security import login_required, current_user	3✔
10	from wtforms import StringField, DecimalField, SelectField	3✔
11	from wtforms.validators import DataRequired, optional	3✔
12	from sqlalchemy import select	3✔
13	from flexmeasures.auth.policy import user_has_admin_access	3✔
14
15	from flexmeasures.data import db	3✔
16	from flexmeasures.auth.error_handling import unauthorized_handler	3✔
17	from flexmeasures.auth.policy import check_access	3✔
18	from flexmeasures.data.models.generic_assets import (	3✔
19	GenericAssetType,
20	GenericAsset,
21	get_center_location_of_assets,
22	)
23	from flexmeasures.data.models.user import Account	3✔
24	from flexmeasures.data.models.time_series import Sensor	3✔
25	from flexmeasures.ui.utils.view_utils import render_flexmeasures_template	3✔
26	from flexmeasures.ui.crud.api_wrapper import InternalApi	3✔
27
28
29	"""	2✔
30	Asset crud view.
31
32	Note: This uses the internal dev API version
33	― if those endpoints get moved or updated to a higher version,
34	we probably should change the version used here, as well.
35	"""
36
37
38	class AssetForm(FlaskForm):	3✔
39	"""The default asset form only allows to edit the name and location."""	3✔
40
41	name = StringField("Name")	3✔
42	latitude = DecimalField(	3✔
43	"Latitude",
44	validators=[optional()],
45	places=None,
46	render_kw={"placeholder": "--Click the map or enter a latitude--"},
47	)
48	longitude = DecimalField(	3✔
49	"Longitude",
50	validators=[optional()],
51	places=None,
52	render_kw={"placeholder": "--Click the map or enter a longitude--"},
53	)
54	attributes = StringField("Other attributes (JSON)", default="{}")	3✔
55
56	def validate_on_submit(self):	3✔
57	if (	3✔
58	hasattr(self, "generic_asset_type_id")
59	and self.generic_asset_type_id.data == -1
60	):
61	self.generic_asset_type_id.data = (	×
62	"" # cannot be coerced to int so will be flagged as invalid input
63	)
64	if hasattr(self, "account_id") and self.account_id.data == -1:	3✔
65	del self.account_id # asset will be public	×
66	return super().validate_on_submit()	3✔
67
68	def to_json(self) -> dict:	3✔
69	"""turn form data into a JSON we can POST to our internal API"""
70	data = copy.copy(self.data)	3✔
71	if data.get("longitude") is not None:	3✔
72	data["longitude"] = float(data["longitude"])	3✔
73	if data.get("latitude") is not None:	3!
74	data["latitude"] = float(data["latitude"])	3✔
75
76	if "csrf_token" in data:	3!
77	del data["csrf_token"]	×
78
79	return data	3✔
80
81	def process_api_validation_errors(self, api_response: dict):	3✔
82	"""Process form errors from the API for the WTForm"""
83	if not isinstance(api_response, dict):	×
84	return	×
85	for error_header in ("json", "validation_errors"):	×
86	if error_header not in api_response:	×
87	continue	×
88	for field in list(self._fields.keys()):	×
89	if field in list(api_response[error_header].keys()):	×
90	field_errors = api_response[error_header][field]	×
91	if isinstance(field_errors, list):	×
92	self._fields[field].errors += api_response[error_header][field]	×
93	else:
94	self._fields[field].errors.append(	×
95	api_response[error_header][field]
96	)
97
98
99	class NewAssetForm(AssetForm):	3✔
100	"""Here, in addition, we allow to set asset type and account."""	3✔
101
102	generic_asset_type_id = SelectField(	3✔
103	"Asset type", coerce=int, validators=[DataRequired()]
104	)
105	account_id = SelectField("Account", coerce=int)	3✔
106
107
108	def with_options(form: AssetForm \| NewAssetForm) -> AssetForm \| NewAssetForm:	3✔
109	if "generic_asset_type_id" in form:	3✔
110	form.generic_asset_type_id.choices = [(-1, "--Select type--")] + [	3✔
111	(atype.id, atype.name)
112	for atype in db.session.scalars(select(GenericAssetType)).all()
113	]
114	if "account_id" in form:	3✔
115	form.account_id.choices = [(-1, "--Select account--")] + [	3✔
116	(account.id, account.name)
117	for account in db.session.scalars(select(Account)).all()
118	]
119	return form	3✔
120
121
122	def process_internal_api_response(	3✔
123	asset_data: dict, asset_id: int \| None = None, make_obj=False
124	) -> GenericAsset \| dict:
125	"""
126	Turn data from the internal API into something we can use to further populate the UI.
127	Either as an asset object or a dict for form filling.
128
129	If we add other data by querying the database, we make sure the asset is not in the session afterwards.
130	"""
131
132	def expunge_asset():	3✔
133	# use if no insert is wanted from a previous query which flushes its results
134	if asset in db.session:	3✔
UNCOV 135	db.session.expunge(asset)	×
136
137	asset_data.pop("status", None) # might have come from requests.response	3✔
138	if asset_id:	3✔
139	asset_data["id"] = asset_id	3✔
140	if make_obj:	3✔
141	asset = GenericAsset(	3✔
142	**{
143	**asset_data,
144	**{"attributes": json.loads(asset_data.get("attributes", "{}"))},
145	}
146	) # TODO: use schema?
147	asset.generic_asset_type = db.session.get(	3✔
148	GenericAssetType, asset.generic_asset_type_id
149	)
150	expunge_asset()	3✔
151	asset.owner = db.session.get(Account, asset_data["account_id"])	3✔
152	expunge_asset()	3✔
153	if "id" in asset_data:	3✔
154	asset.sensors = db.session.scalars(	3✔
155	select(Sensor).filter_by(generic_asset_id=asset_data["id"])
156	).all()
157	expunge_asset()	3✔
158	if asset_data.get("parent_asset_id", None) is not None:	3!
159	asset.parent_asset = GenericAsset.query.filter(	×
160	GenericAsset.id == asset_data["parent_asset_id"]
161	).one_or_none()
162	expunge_asset()	×
163	return asset	3✔
164	return asset_data	3✔
165
166
167	def user_can_create_assets() -> bool:	3✔
168	try:	3✔
169	check_access(current_user.account, "create-children")	3✔
170	except Exception:	3✔
171	return False	3✔
172	return True	3✔
173
174
175	def user_can_delete(asset) -> bool:	3✔
176	try:	3✔
177	check_access(asset, "delete")	3✔
178	except Exception:	3✔
179	return False	3✔
180	return True	3✔
181
182
183	def get_assets_by_account(account_id: int \| str \| None) -> list[GenericAsset]:	3✔
184	if account_id is not None:	3✔
185	get_assets_response = InternalApi().get(	3✔
186	url_for("AssetAPI:index"), query={"account_id": account_id}
187	)
188	else:
189	get_assets_response = InternalApi().get(url_for("AssetAPI:public"))	3✔
190	return [	3✔
191	process_internal_api_response(ad, make_obj=True)
192	for ad in get_assets_response.json()
193	]
194
195
196	class AssetCrudUI(FlaskView):	3✔
197	"""	3✔
198	These views help us offer a Jinja2-based UI.
199	The main focus on logic is the API, so these views simply call the API functions,
200	and deal with the response.
201	Some new functionality, like fetching accounts and asset types, is added here.
202	"""
203
204	route_base = "/assets"	3✔
205
206	@login_required	3✔
207	def index(self, msg=""):	3✔
208	"""GET from /assets
209
210	List the user's assets. For admins, list across all accounts.
211	"""
212	assets = []	3✔
213
214	if user_has_admin_access(current_user, "read"):	3✔
215	for account in db.session.scalars(select(Account)).all():	3✔
216	assets += get_assets_by_account(account.id)	3✔
217	assets += get_assets_by_account(account_id=None)	3✔
218	else:
219	assets = get_assets_by_account(current_user.account_id)	3✔
220
221	return render_flexmeasures_template(	3✔
222	"crud/assets.html",
223	assets=assets,
224	message=msg,
225	user_can_create_assets=user_can_create_assets(),
226	)
227
228	@login_required	3✔
229	def owned_by(self, account_id: str):	3✔
230	"""/assets/owned_by/<account_id>"""
231	msg = ""	3✔
232	get_assets_response = InternalApi().get(	3✔
233	url_for("AssetAPI:index"),
234	query={"account_id": account_id},
235	do_not_raise_for=[404],
236	)
237	if get_assets_response.status_code == 404:	3✔
238	assets = []	×
239	msg = f"Account {account_id} unknown."	×
240	else:
241	assets = [	3✔
242	process_internal_api_response(ad, make_obj=True)
243	for ad in get_assets_response.json()
244	]
245	return render_flexmeasures_template(	3✔
246	"crud/assets.html",
247	account=db.session.get(Account, account_id),
248	assets=assets,
249	msg=msg,
250	user_can_create_assets=user_can_create_assets(),
251	)
252
253	@login_required	3✔
254	def get(self, id: str):	3✔
255	"""GET from /assets/<id> where id can be 'new' (and thus the form for asset creation is shown)"""
256
257	if id == "new":	3✔
258	if not user_can_create_assets():	3✔
259	return unauthorized_handler(None, [])	×
260
261	asset_form = with_options(NewAssetForm())	3✔
262	return render_flexmeasures_template(	3✔
263	"crud/asset_new.html",
264	asset_form=asset_form,
265	msg="",
266	map_center=get_center_location_of_assets(user=current_user),
267	mapboxAccessToken=current_app.config.get("MAPBOX_ACCESS_TOKEN", ""),
268	)
269
270	get_asset_response = InternalApi().get(url_for("AssetAPI:fetch_one", id=id))	3✔
271	asset_dict = get_asset_response.json()	3✔
272
273	asset_form = with_options(AssetForm())	3✔
274
275	asset = process_internal_api_response(asset_dict, int(id), make_obj=True)	3✔
276	asset_form.process(data=process_internal_api_response(asset_dict))	3✔
277
278	return render_flexmeasures_template(	3✔
279	"crud/asset.html",
280	asset=asset,
281	asset_form=asset_form,
282	msg="",
283	mapboxAccessToken=current_app.config.get("MAPBOX_ACCESS_TOKEN", ""),
284	user_can_create_assets=user_can_create_assets(),
285	user_can_delete_asset=user_can_delete(asset),
286	)
287
288	@login_required	3✔
289	def post(self, id: str):	3✔
290	"""POST to /assets/<id>, where id can be 'create' (and thus a new asset is made from POST data)
291	Most of the code deals with creating a user for the asset if no existing is chosen.
292	"""
293
294	asset: GenericAsset = None	3✔
295	error_msg = ""	3✔
296
297	if id == "create":	3✔
298	asset_form = with_options(NewAssetForm())	3✔
299
300	account, account_error = _set_account(asset_form)	3✔
301	asset_type, asset_type_error = _set_asset_type(asset_form)	3✔
302
303	form_valid = asset_form.validate_on_submit()	3✔
304
305	# Fill up the form with useful errors for the user
306	if account_error is not None:	3✔
307	form_valid = False	×
308	asset_form.account_id.errors.append(account_error)	×
309	if asset_type_error is not None:	3✔
310	form_valid = False	×
311	asset_form.generic_asset_type_id.errors.append(asset_type_error)	×
312
313	# Create new asset or return the form for new assets with a message
314	if form_valid and asset_type is not None:	3✔
315	post_asset_response = InternalApi().post(	3✔
316	url_for("AssetAPI:post"),
317	args=asset_form.to_json(),
318	do_not_raise_for=[400, 422],
319	)
320	if post_asset_response.status_code in (200, 201):	3✔
321	asset_dict = post_asset_response.json()	3✔
322	asset = process_internal_api_response(	3✔
323	asset_dict, int(asset_dict["id"]), make_obj=True
324	)
325	msg = "Creation was successful."	3✔
326	else:
327	current_app.logger.error(	×
328	f"Internal asset API call unsuccessful [{post_asset_response.status_code}]: {post_asset_response.text}"
329	)
330	asset_form.process_api_validation_errors(post_asset_response.json())	×
331	if "message" in post_asset_response.json():	×
332	asset_form.process_api_validation_errors(	×
333	post_asset_response.json()["message"]
334	)
335	if "json" in post_asset_response.json()["message"]:	×
336	error_msg = str(	×
337	post_asset_response.json()["message"]["json"]
338	)
339	if asset is None:	3!
340	msg = "Cannot create asset. " + error_msg	×
341	return render_flexmeasures_template(	×
342	"crud/asset_new.html",
343	asset_form=asset_form,
344	msg=msg,
345	map_center=get_center_location_of_assets(user=current_user),
346	mapboxAccessToken=current_app.config.get("MAPBOX_ACCESS_TOKEN", ""),
347	)
348
349	else:
350	asset_form = with_options(AssetForm())	3✔
351	if not asset_form.validate_on_submit():	3✔
NEW 352	asset = db.session.get(GenericAsset, id)	×
353	# Display the form data, but set some extra data which the page wants to show.
354	asset_info = asset_form.to_json()	×
355	asset_info["id"] = id	×
356	asset_info["account_id"] = asset.account_id	×
357	asset = process_internal_api_response(	×
358	asset_info, int(id), make_obj=True
359	)
360
361	return render_flexmeasures_template(	×
362	"crud/asset.html",
363	asset_form=asset_form,
364	asset=asset,
365	msg="Cannot edit asset.",
366	mapboxAccessToken=current_app.config.get("MAPBOX_ACCESS_TOKEN", ""),
367	user_can_create_assets=user_can_create_assets(),
368	user_can_delete_asset=user_can_delete(asset),
369	)
370	patch_asset_response = InternalApi().patch(	3✔
371	url_for("AssetAPI:patch", id=id),
372	args=asset_form.to_json(),
373	do_not_raise_for=[400, 422],
374	)
375	asset_dict = patch_asset_response.json()	3✔
376	if patch_asset_response.status_code in (200, 201):	3✔
377	asset = process_internal_api_response(	3✔
378	asset_dict, int(id), make_obj=True
379	)
380	msg = "Editing was successful."	3✔
381	else:
382	current_app.logger.error(	×
383	f"Internal asset API call unsuccessful [{patch_asset_response.status_code}]: {patch_asset_response.text}"
384	)
385	msg = "Cannot edit asset."	×
386	asset_form.process_api_validation_errors(	×
387	patch_asset_response.json().get("message")
388	)
NEW 389	asset = db.session.get(GenericAsset, id)	×
390
391	return render_flexmeasures_template(	3✔
392	"crud/asset.html",
393	asset=asset,
394	asset_form=asset_form,
395	msg=msg,
396	mapboxAccessToken=current_app.config.get("MAPBOX_ACCESS_TOKEN", ""),
397	user_can_create_assets=user_can_create_assets(),
398	user_can_delete_asset=user_can_delete(asset),
399	)
400
401	@login_required	3✔
402	def delete_with_data(self, id: str):	3✔
403	"""Delete via /assets/delete_with_data/<id>"""
404	InternalApi().delete(url_for("AssetAPI:delete", id=id))	3✔
405	return self.index(	3✔
406	msg=f"Asset {id} and assorted meter readings / forecasts have been deleted."
407	)
408
409
410	def _set_account(asset_form: NewAssetForm) -> tuple[Account \| None, str \| None]:	3✔
411	"""Set an account for the to-be-created asset.
412	Return the account (if available) and an error message"""
413	account_error = None	3✔
414
415	if asset_form.account_id.data == -1:	3✔
416	if user_has_admin_access(current_user, "update"):	×
417	return None, None # Account can be None (public asset)	×
418	else:
419	account_error = "Please pick an existing account."	×
420
421	account = db.session.execute(	3✔
422	select(Account).filter_by(id=int(asset_form.account_id.data))
423	).scalar_one_or_none()
424
425	if account:	3✔
426	asset_form.account_id.data = account.id	3✔
427	else:
428	current_app.logger.error(account_error)	×
429	return account, account_error	3✔
430
431
432	def _set_asset_type(	3✔
433	asset_form: NewAssetForm,
434	) -> tuple[GenericAssetType \| None, str \| None]:
435	"""Set an asset type for the to-be-created asset.
436	Return the asset type (if available) and an error message."""
437	asset_type = None	3✔
438	asset_type_error = None	3✔
439
440	if int(asset_form.generic_asset_type_id.data) == -1:	3✔
441	asset_type_error = "Pick an existing asset type."	×
442	else:
443	asset_type = db.session.execute(	3✔
444	select(GenericAssetType).filter_by(
445	id=int(asset_form.generic_asset_type_id.data)
446	)
447	).scalar_one_or_none()
448
449	if asset_type:	3✔
450	asset_form.generic_asset_type_id.data = asset_type.id	3✔
451	else:
452	current_app.logger.error(asset_type_error)	×
453	return asset_type, asset_type_error	3✔

FlexMeasures / flexmeasures / 7528577372

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