10850704497

Committed 13 Sep 2024 02:11PM UTC coverage: 67.038% (-0.4%) from 67.402%

Build # 10850704497

Build Type

Pull #1118

github

Committed by

Flix6x

Commit Message

dev: workaround to select most recent version of sources representing the same model

Signed-off-by: F.N. Claessen <felix@seita.nl>

Pull Request Pull Request #1118: Feature/planning/soft site power capacities

Run Details

6245 of 10162 branches covered (61.45%)

Branch coverage included in aggregate %.

378 of 549 new or added lines in 36 files covered. (68.85%)

13 existing lines in 7 files now uncovered.

7908 of 10950 relevant lines covered (72.22%)

5.72 hits per line

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

90.64

/flexmeasures/data/models/user.py

from __future__ import annotations

from typing import TYPE_CHECKING
from datetime import datetime

from flask_security import UserMixin, RoleMixin
import pandas as pd
from sqlalchemy import select, func
from sqlalchemy.orm import relationship, backref
from sqlalchemy import Boolean, DateTime, Column, Integer, String, ForeignKey
from sqlalchemy.ext.hybrid import hybrid_property
from timely_beliefs import utils as tb_utils

from flexmeasures.data import db
from flexmeasures.data.models.annotations import (
    Annotation,
    AccountAnnotationRelationship,
    to_annotation_frame,
)
from flexmeasures.data.models.parsing_utils import parse_source_arg
from flexmeasures.auth.policy import AuthModelMixin

if TYPE_CHECKING:
    from flexmeasures.data.models.data_sources import DataSource


class RolesAccounts(db.Model):
    __tablename__ = "roles_accounts"
    id = Column(Integer(), primary_key=True)
    account_id = Column("account_id", Integer(), ForeignKey("account.id"))
    role_id = Column("role_id", Integer(), ForeignKey("account_role.id"))
    __table_args__ = (
        db.UniqueConstraint(
            "role_id",
            "account_id",
            name="roles_accounts_role_id_key",
        ),
    )


class AccountRole(db.Model):
    __tablename__ = "account_role"
    id = Column(Integer(), primary_key=True)
    name = Column(String(80), unique=True)
    description = Column(String(255))

    def __repr__(self):
        return "<AccountRole:%s (ID:%s)>" % (self.name, self.id)


class Account(db.Model, AuthModelMixin):
    """
    Account of a tenant on the server.
    Bundles Users as well as GenericAssets.
    """

    __tablename__ = "account"
    id = Column(Integer, primary_key=True)
    name = Column(String(100), default="", unique=True)
    account_roles = relationship(
        "AccountRole",
        secondary="roles_accounts",
        backref=backref("accounts", lazy="dynamic"),
    )
    primary_color = Column(String(7), default=None)
    secondary_color = Column(String(7), default=None)
    logo_url = Column(String(255), default=None)
    annotations = db.relationship(
        "Annotation",
        secondary="annotations_accounts",
        backref=db.backref("accounts", lazy="dynamic"),
    )

    # Setup self-referential relationship between consultancy account and consultancy client account
    consultancy_account_id = Column(
        Integer, db.ForeignKey("account.id"), default=None, nullable=True
    )
    consultancy_client_accounts = db.relationship(
        "Account", back_populates="consultancy_account"
    )
    consultancy_account = db.relationship(
        "Account", back_populates="consultancy_client_accounts", remote_side=[id]
    )

    def __repr__(self):
        return "<Account %s (ID:%s)>" % (self.name, self.id)

    def __acl__(self):
        """
        Only account admins can create things in the account (e.g. users or assets).
        Consultants (i.e. users with the consultant role) can read things in the account,
        but only if their organisation is set as a consultancy for the given account.
        Within same account, everyone can read and update.
        Creation and deletion of accounts are left to site admins in CLI.
        """

        read_access = [f"account:{self.id}"]
        if self.consultancy_account_id is not None:
            read_access.append(
                (f"account:{self.consultancy_account_id}", "role:consultant")
            )
        return {
            "create-children": (f"account:{self.id}", "role:account-admin"),
            "read": read_access,
            "update": f"account:{self.id}",
        }

    def get_path(self, separator: str = ">"):
        return self.name

    def has_role(self, role: str | AccountRole) -> bool:
        """Returns `True` if the account has the specified role.

        :param role: An account role name or `AccountRole` instance"""
        if isinstance(role, str):
            return role in (role.name for role in self.account_roles)
        else:
            return role in self.account_roles

    def search_annotations(
        self,
        annotation_starts_after: datetime | None = None,  # deprecated
        annotations_after: datetime | None = None,
        annotation_ends_before: datetime | None = None,  # deprecated
        annotations_before: datetime | None = None,
        source: (
            DataSource | list[DataSource] | int | list[int] | str | list[str] | None
        ) = None,
        as_frame: bool = False,
    ) -> list[Annotation] | pd.DataFrame:
        """Return annotations assigned to this account.

        :param annotations_after: only return annotations that end after this datetime (exclusive)
        :param annotations_before: only return annotations that start before this datetime (exclusive)
        """

        # todo: deprecate the 'annotation_starts_after' argument in favor of 'annotations_after' (announced v0.11.0)
        annotations_after = tb_utils.replace_deprecated_argument(
            "annotation_starts_after",
            annotation_starts_after,
            "annotations_after",
            annotations_after,
            required_argument=False,
        )

        # todo: deprecate the 'annotation_ends_before' argument in favor of 'annotations_before' (announced v0.11.0)
        annotations_before = tb_utils.replace_deprecated_argument(
            "annotation_ends_before",
            annotation_ends_before,
            "annotations_before",
            annotations_before,
            required_argument=False,
        )

        parsed_sources = parse_source_arg(source)
        query = (
            select(Annotation)
            .join(AccountAnnotationRelationship)
            .filter(
                AccountAnnotationRelationship.account_id == self.id,
                AccountAnnotationRelationship.annotation_id == Annotation.id,
            )
        )
        if annotations_after is not None:
            query = query.filter(
                Annotation.end > annotations_after,
            )
        if annotations_before is not None:
            query = query.filter(
                Annotation.start < annotations_before,
            )
        if parsed_sources:
            query = query.filter(
                Annotation.source.in_(parsed_sources),
            )
        annotations = db.session.scalars(query).all()

        return to_annotation_frame(annotations) if as_frame else annotations

    @property
    def number_of_assets(self):
        from flexmeasures.data.models.generic_assets import GenericAsset

        return db.session.execute(
            select(func.count()).where(GenericAsset.account_id == self.id)
        ).scalar_one_or_none()

    @property
    def number_of_users(self):
        return db.session.execute(
            select(func.count()).where(User.account_id == self.id)
        ).scalar_one_or_none()


class RolesUsers(db.Model):
    __tablename__ = "roles_users"
    id = Column(Integer(), primary_key=True)
    user_id = Column("user_id", Integer(), ForeignKey("fm_user.id"))
    role_id = Column("role_id", Integer(), ForeignKey("role.id"))
    __table_args__ = (
        db.UniqueConstraint(
            "role_id",
            "user_id",
            name="roles_users_role_id_key",
        ),
    )


class Role(db.Model, RoleMixin):
    __tablename__ = "role"
    id = Column(Integer(), primary_key=True)
    name = Column(String(80), unique=True)
    description = Column(String(255))

    def __repr__(self):
        return "<Role:%s (ID:%s)>" % (self.name, self.id)


class User(db.Model, UserMixin, AuthModelMixin):
    """
    We use the flask security UserMixin, which does include functionality,
    but not the fields (those are in flask_security/models::FsUserMixin).
    We went with a pick&choose approach. This gives us more freedom, e.g.
    to choose our own table name or add logic around the activation status.
    If we add new FS functionality (e.g. 2FA), the fields needed for that
    need to be added here.
    """

    __tablename__ = "fm_user"

    id = Column(Integer, primary_key=True)
    email = Column(String(255), unique=True)
    username = Column(String(255), unique=True)
    password = Column(String(255))
    # Last time the user logged in (provided credentials to get access)
    last_login_at = Column(DateTime())
    # Last time the user made a request (authorized by session or auth token)
    last_seen_at = Column(DateTime())
    # How often have they logged in?
    login_count = Column(Integer)
    active = Column(Boolean())
    # Faster token checking
    fs_uniquifier = Column(String(64), unique=True, nullable=False)
    timezone = Column(String(255), default="Europe/Amsterdam")
    account_id = Column(Integer, db.ForeignKey("account.id"), nullable=False)

    account = db.relationship("Account", backref=db.backref("users", lazy=True))
    flexmeasures_roles = relationship(
        "Role",
        secondary="roles_users",
        backref=backref("users", lazy="dynamic"),
    )

    def __repr__(self):
        return "<User %s (ID:%s)>" % (self.username, self.id)

    def __acl__(self):
        """
        Within same account, everyone can read.
        Only the user themselves or account-admins can edit their user record.
        Creation and deletion are left to site admins in CLI.
        """
        return {
            "read": f"account:{self.account_id}",
            "update": [
                f"user:{self.id}",
                (f"account:{self.account_id}", "role:account-admin"),
            ],
        }

    @property
    def is_authenticated(self) -> bool:
        """We are overloading this, so it also considers being active.
        Inactive users can by definition not be authenticated."""
        return super(UserMixin, self).is_authenticated and self.active

    @hybrid_property
    def roles(self):
        """The roles attribute is being used by Flask-Security in the @roles_required decorator (among others).
        With this little overload fix, it will only return the user's roles if they are authenticated.
        We do this to prevent that if a user is logged in while the admin deactivates them, their session would still work.
        In effect, we strip unauthenticated users from their roles. To read roles of an unauthenticated user
        (e.g. being inactive), use the `flexmeasures_roles` attribute.
        If our auth model has moved to an improved way, e.g. requiring modern tokens, we should consider relaxing this.
        Note: This needed to become a hybrid property when moving to Flask-Security 3.4
        """
        if not self.is_authenticated and self is not User:
            return []
        else:
            return self.flexmeasures_roles

    @roles.setter
    def roles(self, new_roles):
        """See comment in roles property why we overload."""
        self.flexmeasures_roles = new_roles

    def has_role(self, role: str | Role) -> bool:
        """Returns `True` if the user identifies with the specified role.
            Overwritten from flask_security.core.UserMixin.

        :param role: A role name or `Role` instance"""
        if isinstance(role, str):
            return role in (role.name for role in self.flexmeasures_roles)
        else:
            return role in self.flexmeasures_roles


def remember_login(the_app, user):
    """We do not use the tracking feature of flask_security, but this basic meta data are quite handy to know"""
    user.last_login_at = datetime.utcnow()
    if user.login_count is None:
        user.login_count = 0
    user.login_count = user.login_count + 1


def remember_last_seen(user):
    """Update the last_seen field"""
    if user is not None and user.is_authenticated:
        user.last_seen_at = datetime.utcnow()
        db.session.add(user)
        db.session.commit()


def is_user(o) -> bool:
    """True if object is or proxies a User, False otherwise.

    Takes into account that object can be of LocalProxy type, and
    uses get_current_object to get the underlying (User) object.
    """
    return isinstance(o, User) or (
        hasattr(o, "_get_current_object") and isinstance(o._get_current_object(), User)
    )

1	from __future__ import annotations	8✔
2
3	from typing import TYPE_CHECKING	8✔
4	from datetime import datetime	8✔
5
6	from flask_security import UserMixin, RoleMixin	8✔
7	import pandas as pd	8✔
8	from sqlalchemy import select, func	8✔
9	from sqlalchemy.orm import relationship, backref	8✔
10	from sqlalchemy import Boolean, DateTime, Column, Integer, String, ForeignKey	8✔
11	from sqlalchemy.ext.hybrid import hybrid_property	8✔
12	from timely_beliefs import utils as tb_utils	8✔
13
14	from flexmeasures.data import db	8✔
15	from flexmeasures.data.models.annotations import (	8✔
16	Annotation,
17	AccountAnnotationRelationship,
18	to_annotation_frame,
19	)
20	from flexmeasures.data.models.parsing_utils import parse_source_arg	8✔
21	from flexmeasures.auth.policy import AuthModelMixin	8✔
22
23	if TYPE_CHECKING:	8✔
24	from flexmeasures.data.models.data_sources import DataSource	×
25
26
27	class RolesAccounts(db.Model):	8✔
28	__tablename__ = "roles_accounts"	8✔
29	id = Column(Integer(), primary_key=True)	8✔
30	account_id = Column("account_id", Integer(), ForeignKey("account.id"))	8✔
31	role_id = Column("role_id", Integer(), ForeignKey("account_role.id"))	8✔
32	__table_args__ = (	8✔
33	db.UniqueConstraint(
34	"role_id",
35	"account_id",
36	name="roles_accounts_role_id_key",
37	),
38	)
39
40
41	class AccountRole(db.Model):	8✔
42	__tablename__ = "account_role"	8✔
43	id = Column(Integer(), primary_key=True)	8✔
44	name = Column(String(80), unique=True)	8✔
45	description = Column(String(255))	8✔
46
47	def __repr__(self):	8✔
48	return "<AccountRole:%s (ID:%s)>" % (self.name, self.id)	×
49
50
51	class Account(db.Model, AuthModelMixin):	8✔
52	"""	4✔
53	Account of a tenant on the server.
54	Bundles Users as well as GenericAssets.
55	"""
56
57	__tablename__ = "account"	8✔
58	id = Column(Integer, primary_key=True)	8✔
59	name = Column(String(100), default="", unique=True)	8✔
60	account_roles = relationship(	8✔
61	"AccountRole",
62	secondary="roles_accounts",
63	backref=backref("accounts", lazy="dynamic"),
64	)
65	primary_color = Column(String(7), default=None)	8✔
66	secondary_color = Column(String(7), default=None)	8✔
67	logo_url = Column(String(255), default=None)	8✔
68	annotations = db.relationship(	8✔
69	"Annotation",
70	secondary="annotations_accounts",
71	backref=db.backref("accounts", lazy="dynamic"),
72	)
73
74	# Setup self-referential relationship between consultancy account and consultancy client account
75	consultancy_account_id = Column(	8✔
76	Integer, db.ForeignKey("account.id"), default=None, nullable=True
77	)
78	consultancy_client_accounts = db.relationship(	8✔
79	"Account", back_populates="consultancy_account"
80	)
81	consultancy_account = db.relationship(	8✔
82	"Account", back_populates="consultancy_client_accounts", remote_side=[id]
83	)
84
85	def __repr__(self):	8✔
86	return "<Account %s (ID:%s)>" % (self.name, self.id)	8✔
87
88	def __acl__(self):	8✔
89	"""
90	Only account admins can create things in the account (e.g. users or assets).
91	Consultants (i.e. users with the consultant role) can read things in the account,
92	but only if their organisation is set as a consultancy for the given account.
93	Within same account, everyone can read and update.
94	Creation and deletion of accounts are left to site admins in CLI.
95	"""
96
97	read_access = [f"account:{self.id}"]	8✔
98	if self.consultancy_account_id is not None:	8✔
99	read_access.append(	8✔
100	(f"account:{self.consultancy_account_id}", "role:consultant")
101	)
102	return {	8✔
103	"create-children": (f"account:{self.id}", "role:account-admin"),
104	"read": read_access,
105	"update": f"account:{self.id}",
106	}
107
108	def get_path(self, separator: str = ">"):	8✔
109	return self.name	8✔
110
111	def has_role(self, role: str \| AccountRole) -> bool:	8✔
112	"""Returns `True` if the account has the specified role.
113
114	:param role: An account role name or `AccountRole` instance"""
115	if isinstance(role, str):	8✔
116	return role in (role.name for role in self.account_roles)	8✔
117	else:
118	return role in self.account_roles	×
119
120	def search_annotations(	8✔
121	self,
122	annotation_starts_after: datetime \| None = None, # deprecated
123	annotations_after: datetime \| None = None,
124	annotation_ends_before: datetime \| None = None, # deprecated
125	annotations_before: datetime \| None = None,
126	source: (
127	DataSource \| list[DataSource] \| int \| list[int] \| str \| list[str] \| None
128	) = None,
129	as_frame: bool = False,
130	) -> list[Annotation] \| pd.DataFrame:
131	"""Return annotations assigned to this account.
132
133	:param annotations_after: only return annotations that end after this datetime (exclusive)
134	:param annotations_before: only return annotations that start before this datetime (exclusive)
135	"""
136
137	# todo: deprecate the 'annotation_starts_after' argument in favor of 'annotations_after' (announced v0.11.0)
138	annotations_after = tb_utils.replace_deprecated_argument(	8✔
139	"annotation_starts_after",
140	annotation_starts_after,
141	"annotations_after",
142	annotations_after,
143	required_argument=False,
144	)
145
146	# todo: deprecate the 'annotation_ends_before' argument in favor of 'annotations_before' (announced v0.11.0)
147	annotations_before = tb_utils.replace_deprecated_argument(	8✔
148	"annotation_ends_before",
149	annotation_ends_before,
150	"annotations_before",
151	annotations_before,
152	required_argument=False,
153	)
154
155	parsed_sources = parse_source_arg(source)	8✔
156	query = (	8✔
157	select(Annotation)
158	.join(AccountAnnotationRelationship)
159	.filter(
160	AccountAnnotationRelationship.account_id == self.id,
161	AccountAnnotationRelationship.annotation_id == Annotation.id,
162	)
163	)
164	if annotations_after is not None:	8✔
165	query = query.filter(	×
166	Annotation.end > annotations_after,
167	)
168	if annotations_before is not None:	8✔
169	query = query.filter(	×
170	Annotation.start < annotations_before,
171	)
172	if parsed_sources:	8✔
173	query = query.filter(	×
174	Annotation.source.in_(parsed_sources),
175	)
176	annotations = db.session.scalars(query).all()	8✔
177
178	return to_annotation_frame(annotations) if as_frame else annotations	8✔
179
180	@property	8✔
181	def number_of_assets(self):	8✔
NEW 182	from flexmeasures.data.models.generic_assets import GenericAsset	×
183
NEW 184	return db.session.execute(	×
185	select(func.count()).where(GenericAsset.account_id == self.id)
186	).scalar_one_or_none()
187
188	@property	8✔
189	def number_of_users(self):	8✔
NEW 190	return db.session.execute(	×
191	select(func.count()).where(User.account_id == self.id)
192	).scalar_one_or_none()
193
194
195	class RolesUsers(db.Model):	8✔
196	__tablename__ = "roles_users"	8✔
197	id = Column(Integer(), primary_key=True)	8✔
198	user_id = Column("user_id", Integer(), ForeignKey("fm_user.id"))	8✔
199	role_id = Column("role_id", Integer(), ForeignKey("role.id"))	8✔
200	__table_args__ = (	8✔
201	db.UniqueConstraint(
202	"role_id",
203	"user_id",
204	name="roles_users_role_id_key",
205	),
206	)
207
208
209	class Role(db.Model, RoleMixin):	8✔
210	__tablename__ = "role"	8✔
211	id = Column(Integer(), primary_key=True)	8✔
212	name = Column(String(80), unique=True)	8✔
213	description = Column(String(255))	8✔
214
215	def __repr__(self):	8✔
216	return "<Role:%s (ID:%s)>" % (self.name, self.id)	×
217
218
219	class User(db.Model, UserMixin, AuthModelMixin):	8✔
220	"""	4✔
221	We use the flask security UserMixin, which does include functionality,
222	but not the fields (those are in flask_security/models::FsUserMixin).
223	We went with a pick&choose approach. This gives us more freedom, e.g.
224	to choose our own table name or add logic around the activation status.
225	If we add new FS functionality (e.g. 2FA), the fields needed for that
226	need to be added here.
227	"""
228
229	__tablename__ = "fm_user"	8✔
230
231	id = Column(Integer, primary_key=True)	8✔
232	email = Column(String(255), unique=True)	8✔
233	username = Column(String(255), unique=True)	8✔
234	password = Column(String(255))	8✔
235	# Last time the user logged in (provided credentials to get access)
236	last_login_at = Column(DateTime())	8✔
237	# Last time the user made a request (authorized by session or auth token)
238	last_seen_at = Column(DateTime())	8✔
239	# How often have they logged in?
240	login_count = Column(Integer)	8✔
241	active = Column(Boolean())	8✔
242	# Faster token checking
243	fs_uniquifier = Column(String(64), unique=True, nullable=False)	8✔
244	timezone = Column(String(255), default="Europe/Amsterdam")	8✔
245	account_id = Column(Integer, db.ForeignKey("account.id"), nullable=False)	8✔
246
247	account = db.relationship("Account", backref=db.backref("users", lazy=True))	8✔
248	flexmeasures_roles = relationship(	8✔
249	"Role",
250	secondary="roles_users",
251	backref=backref("users", lazy="dynamic"),
252	)
253
254	def __repr__(self):	8✔
255	return "<User %s (ID:%s)>" % (self.username, self.id)	8✔
256
257	def __acl__(self):	8✔
258	"""
259	Within same account, everyone can read.
260	Only the user themselves or account-admins can edit their user record.
261	Creation and deletion are left to site admins in CLI.
262	"""
263	return {	8✔
264	"read": f"account:{self.account_id}",
265	"update": [
266	f"user:{self.id}",
267	(f"account:{self.account_id}", "role:account-admin"),
268	],
269	}
270
271	@property	8✔
272	def is_authenticated(self) -> bool:	8✔
273	"""We are overloading this, so it also considers being active.
274	Inactive users can by definition not be authenticated."""
275	return super(UserMixin, self).is_authenticated and self.active	8✔
276
277	@hybrid_property	8✔
278	def roles(self):	8✔
279	"""The roles attribute is being used by Flask-Security in the @roles_required decorator (among others).
280	With this little overload fix, it will only return the user's roles if they are authenticated.
281	We do this to prevent that if a user is logged in while the admin deactivates them, their session would still work.
282	In effect, we strip unauthenticated users from their roles. To read roles of an unauthenticated user
283	(e.g. being inactive), use the `flexmeasures_roles` attribute.
284	If our auth model has moved to an improved way, e.g. requiring modern tokens, we should consider relaxing this.
285	Note: This needed to become a hybrid property when moving to Flask-Security 3.4
286	"""
287	if not self.is_authenticated and self is not User:	8✔
288	return []	8✔
289	else:
290	return self.flexmeasures_roles	8✔
291
292	@roles.setter	8✔
293	def roles(self, new_roles):	8✔
294	"""See comment in roles property why we overload."""
295	self.flexmeasures_roles = new_roles	8✔
296
297	def has_role(self, role: str \| Role) -> bool:	8✔
298	"""Returns `True` if the user identifies with the specified role.
299	Overwritten from flask_security.core.UserMixin.
300
301	:param role: A role name or `Role` instance"""
302	if isinstance(role, str):	8✔
303	return role in (role.name for role in self.flexmeasures_roles)	8✔
304	else:
305	return role in self.flexmeasures_roles	×
306
307
308	def remember_login(the_app, user):	8✔
309	"""We do not use the tracking feature of flask_security, but this basic meta data are quite handy to know"""
310	user.last_login_at = datetime.utcnow()	8✔
311	if user.login_count is None:	8✔
312	user.login_count = 0	8✔
313	user.login_count = user.login_count + 1	8✔
314
315
316	def remember_last_seen(user):	8✔
317	"""Update the last_seen field"""
318	if user is not None and user.is_authenticated:	8✔
319	user.last_seen_at = datetime.utcnow()	8✔
320	db.session.add(user)	8✔
321	db.session.commit()	8✔
322
323
324	def is_user(o) -> bool:	8✔
325	"""True if object is or proxies a User, False otherwise.
326
327	Takes into account that object can be of LocalProxy type, and
328	uses get_current_object to get the underlying (User) object.
329	"""
330	return isinstance(o, User) or (	8✔
331	hasattr(o, "_get_current_object") and isinstance(o._get_current_object(), User)
332	)

FlexMeasures / flexmeasures / 10850704497

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