"""
Models of gn_notifications schema
"""
import datetime
from math import perm
from geonature.core.gn_commons.models.base import TModules
from geonature.core.gn_permissions.models import PermAction, PermObject
from geonature.core.gn_permissions.tools import get_user_permissions
import sqlalchemy as sa
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
from flask import g
from utils_flask_sqla.models import qfilter
from utils_flask_sqla.serializers import serializable
from pypnusershub.db.models import User
from geonature.utils.env import db
@serializable
[docs]
class NotificationMethod(db.Model):
[docs]
__tablename__ = "bib_notifications_methods"
[docs]
__table_args__ = {"schema": "gn_notifications"}
[docs]
code = db.Column(db.Unicode, primary_key=True)
[docs]
label = db.Column(db.Unicode)
[docs]
description = db.Column(db.UnicodeText)
@property
[docs]
def display(self):
if self.label:
return f"{self} – {self.label}"
else:
return str(self)
[docs]
def __str__(self):
return self.code
@serializable
[docs]
class NotificationCategory(db.Model):
[docs]
__tablename__ = "bib_notifications_categories"
[docs]
__table_args__ = {"schema": "gn_notifications"}
[docs]
code = db.Column(db.Unicode, primary_key=True)
[docs]
label = db.Column(db.Unicode)
[docs]
description = db.Column(db.UnicodeText)
[docs]
id_module = db.Column(db.Integer, ForeignKey("gn_commons.t_modules.id_module"))
[docs]
module = relationship(TModules)
[docs]
id_object = db.Column(db.Integer, ForeignKey("gn_permissions.t_objects.id_object"))
[docs]
object = relationship(PermObject)
[docs]
id_action = db.Column(db.Integer, ForeignKey("gn_permissions.bib_actions.id_action"))
[docs]
action = relationship(PermAction)
[docs]
def is_allowed(self, user=None) -> bool:
if user is None:
user = g.current_user
id_role = user.id_role
permissions = get_user_permissions(id_role)
if self.id_module:
permissions = [p for p in permissions if p.id_module == self.id_module]
if self.id_object:
permissions = [p for p in permissions if p.id_object == self.id_object]
if self.id_action:
permissions = [p for p in permissions if p.id_action == self.id_action]
return bool(permissions)
@property
[docs]
def display(self):
if self.label:
return f"{self} – {self.label}"
else:
return str(self)
[docs]
def __str__(self):
return self.code
@serializable
[docs]
class NotificationTemplate(db.Model):
[docs]
__tablename__ = "bib_notifications_templates"
[docs]
__table_args__ = {"schema": "gn_notifications"}
[docs]
code_category = db.Column(
db.Unicode,
ForeignKey(NotificationCategory.code),
primary_key=True,
)
[docs]
code_method = db.Column(db.Unicode, ForeignKey(NotificationMethod.code), primary_key=True)
[docs]
content = db.Column(db.UnicodeText)
[docs]
category = db.relationship(NotificationCategory)
[docs]
method = db.relationship(NotificationMethod)
[docs]
def __str__(self):
return self.content
@serializable
[docs]
class Notification(db.Model):
[docs]
__tablename__ = "t_notifications"
[docs]
__table_args__ = {"schema": "gn_notifications"}
[docs]
id_notification = db.Column(db.Integer, primary_key=True)
[docs]
id_role = db.Column(db.Integer, ForeignKey(User.id_role), nullable=False)
[docs]
title = db.Column(db.Unicode)
[docs]
content = db.Column(db.UnicodeText)
[docs]
url = db.Column(db.Unicode)
[docs]
code_status = db.Column(db.Unicode)
[docs]
creation_date = db.Column(db.DateTime(), default=datetime.datetime.utcnow)
[docs]
user = db.relationship(User)
@serializable
[docs]
class NotificationRule(db.Model):
[docs]
__tablename__ = "t_notifications_rules"
[docs]
__table_args__ = (
db.UniqueConstraint(
"id_role", "code_method", "code_category", name="un_role_method_category"
),
db.Index(
"un_method_category",
"code_method",
"code_category",
unique=True,
postgresql_ops={
"where": sa.text("id_role IS NULL"),
},
),
{"schema": "gn_notifications"},
)
[docs]
id = db.Column(db.Integer, primary_key=True)
[docs]
id_role = db.Column(db.Integer, ForeignKey(User.id_role), nullable=True)
[docs]
code_method = db.Column(db.Unicode, ForeignKey(NotificationMethod.code), nullable=False)
[docs]
code_category = db.Column(
db.Unicode,
ForeignKey(NotificationCategory.code),
nullable=False,
)
[docs]
subscribed = db.Column(db.Boolean, nullable=False)
[docs]
method = relationship(NotificationMethod)
[docs]
category = relationship(NotificationCategory)
[docs]
user = db.relationship(User)
@qfilter(query=True)
[docs]
def filter_by_role_with_defaults(cls, *, query, id_role=None):
if id_role is None:
id_role = g.current_user.id_role
cte = (
sa.select(NotificationRule)
.where(
sa.or_(
NotificationRule.id_role.is_(None),
NotificationRule.id_role == id_role,
)
)
.distinct(NotificationRule.code_category, NotificationRule.code_method)
.order_by(
NotificationRule.code_category.desc(),
NotificationRule.code_method.desc(),
NotificationRule.id_role.asc(),
)
.cte("cte")
)
return query.where(NotificationRule.id == cte.c.id)