erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

core: notifications implementation, first draft:

Browse source 
add a new table for notifications, and methods/bridge methods to manipulate them.
This commit is contained in:
Goffi 2023-10-16 17:29:31 +02:00
commit b74f55d73d
10 changed files with 1081 additions and 363 deletions
Download patch file Download diff file Expand all files Collapse all files

71
libervia/backend/bridge/bridge_constructor/bridge_template.ini
View file

@ -147,6 +147,31 @@ doc_param_1=name: Name of the updated value
doc_param_2=value: New value
doc_param_3=%(doc_profile)s
[notification_new]
type=signal
category=core
sig_in=sdssssbidss
doc=A new notification has been emitted
doc_param_0=id: unique identifier of the notification
doc_param_1=timestamp: notification creation time
doc_param_2=type: type of the notification
doc_param_3=body_plain: plain text body of the notification
doc_param_4=body_rich: rich text (XHTML) body of the notification. Optional.
doc_param_5=title: optional title of the notification
doc_param_6=requires_action: True if the notification requires user action (e.g. a dialog needs to be answered), False otherwise
doc_param_7=priority: how urgent the notification is, represented as an enumeration value
doc_param_8=expire_at: expiration timestamp for the notification. Optional.
doc_param_9=extra: additional serialized data associated with the notification
doc_param_10=profile: profile associated with the notification. C.PROF_KEY_ALL can be used for global notifications.
[notification_deleted]
type=signal
category=core
sig_in=ss
doc=A new notification has been emitted
doc_param_0=id: id of the deleted notification
doc_param_1=profile: profile of the deleted application, or C.PROF_KEY_ALL for a global notification
[progress_started]
type=signal
category=core
@ -1024,3 +1049,49 @@ doc_param_3=extra: serialised extra
doc_param_4=profile_key: either profile_key or empty string to use common cache
this parameter is used only when dest is empty
doc_return=path to the new converted image
[notification_add]
type=method
category=core
sig_in=ssssbbsdss
sig_out=
doc=Add a new notification
doc_param_0=type_: Notification type
doc_param_1=body_plain: Plain text body of the notification
doc_param_2=body_rich: Rich text body of the notification (optional, can be empty string for default)
doc_param_3=title: Title of the notification (optional, can be empty string for default)
doc_param_4=is_global: True if the notification is for all profiles
doc_param_5=requires_action: Indicates if the notification requires action
doc_param_7=priority: Priority level of the notification (e.g. MEDIUM, HIGH, etc.)
doc_param_8=expire_at: Expiration timestamp for the notification (optional, can be 0 for none)
doc_param_9=extra: Additional details for the notification as a dictionary (optional, can be empty dictionary)
doc_param_10=%(doc_profile_key)s: Profile key (use "@ALL@" for all profiles)
[notifications_get]
type=method
category=core
sig_in=ss
sig_out=s
doc=Retrieve notifications based on provided filters
doc_param_0=filters: a dictionary with filter criteria for notifications retrieval
doc_param_1=%(doc_profile_key)s or @ALL@ for all profiles
doc_return=list of Notification objects. The exact structure will depend on your Notification class.
[notification_delete]
type=method
category=core
sig_in=sbs
sig_out=
doc=Delete a notification
doc_param_0=id_: ID of the notification to delete
doc_param_1=is_global: true if the notification is a global one
doc_param_2=profile_key: Profile key (use "@ALL@" for all profiles)
[notifications_expired_clean]
type=method
category=core
sig_in=ds
sig_out=
doc=Cleans expired notifications and older profile-specific notifications
doc_param_0=limit_timestamp: Timestamp limit for older notifications. If -1.0, only truly expired notifications are removed.
doc_param_1=profile_key: Profile key (use "@NONE@" to indicate no specific profile, otherwise only notification for given profile will be expired, in addition to truly expired notifications).

24
libervia/backend/bridge/dbus_bridge.py
View file

@ -124,6 +124,10 @@ class DBusObject(objects.DBusObject):
Method('message_encryption_stop', arguments='ss', returns=''),
Method('message_send', arguments='sa{ss}a{ss}sss', returns=''),
Method('namespaces_get', arguments='', returns='a{ss}'),
Method('notification_add', arguments='ssssbbsdss', returns=''),
Method('notification_delete', arguments='sbs', returns=''),
Method('notifications_expired_clean', arguments='ds', returns=''),
Method('notifications_get', arguments='ss', returns='s'),
Method('param_get_a', arguments='ssss', returns='s'),
Method('param_get_a_async', arguments='sssis', returns='s'),
Method('param_set', arguments='sssis', returns=''),
@ -164,6 +168,8 @@ class DBusObject(objects.DBusObject):
Signal('message_encryption_started', 'sss'),
Signal('message_encryption_stopped', 'sa{ss}s'),
Signal('message_new', 'sdssa{ss}a{ss}sss'),
Signal('notification_deleted', 'ss'),
Signal('notification_new', 'sdssssbidss'),
Signal('param_update', 'ssss'),
Signal('presence_update', 'ssia{ss}s'),
Signal('progress_error', 'sss'),
@ -304,6 +310,18 @@ class DBusObject(objects.DBusObject):
def dbus_namespaces_get(self, ):
return self._callback("namespaces_get", )
def dbus_notification_add(self, type_, body_plain, body_rich, title, is_global, requires_action, arg_6, priority, expire_at, extra):
return self._callback("notification_add", type_, body_plain, body_rich, title, is_global, requires_action, arg_6, priority, expire_at, extra)
def dbus_notification_delete(self, id_, is_global, profile_key):
return self._callback("notification_delete", id_, is_global, profile_key)
def dbus_notifications_expired_clean(self, limit_timestamp, profile_key):
return self._callback("notifications_expired_clean", limit_timestamp, profile_key)
def dbus_notifications_get(self, filters, profile_key):
return self._callback("notifications_get", filters, profile_key)
def dbus_param_get_a(self, name, category, attribute="value", profile_key="@DEFAULT@"):
return self._callback("param_get_a", name, category, attribute, profile_key)
@ -447,6 +465,12 @@ class bridge:
def message_new(self, uid, timestamp, from_jid, to_jid, message, subject, mess_type, extra, profile):
self._obj.emitSignal("message_new", uid, timestamp, from_jid, to_jid, message, subject, mess_type, extra, profile)
def notification_deleted(self, id, profile):
self._obj.emitSignal("notification_deleted", id, profile)
def notification_new(self, id, timestamp, type, body_plain, body_rich, title, requires_action, priority, expire_at, extra, profile):
self._obj.emitSignal("notification_new", id, timestamp, type, body_plain, body_rich, title, requires_action, priority, expire_at, extra, profile)
def param_update(self, name, value, category, profile):
self._obj.emitSignal("param_update", name, value, category, profile)

6
libervia/backend/bridge/pb.py
View file

@ -193,6 +193,12 @@ class bridge(object):
def message_new(self, uid, timestamp, from_jid, to_jid, message, subject, mess_type, extra, profile):
self.send_signal("message_new", uid, timestamp, from_jid, to_jid, message, subject, mess_type, extra, profile)
def notification_deleted(self, id, profile):
self.send_signal("notification_deleted", id, profile)
def notification_new(self, id, timestamp, type, body_plain, body_rich, title, requires_action, priority, expire_at, extra, profile):
self.send_signal("notification_new", id, timestamp, type, body_plain, body_rich, title, requires_action, priority, expire_at, extra, profile)
def param_update(self, name, value, category, profile):
self.send_signal("param_update", name, value, category, profile)

4
libervia/backend/core/main.py
View file

@ -221,6 +221,10 @@ class LiberviaBackend(service.Service):
self.bridge.register_method("image_resize", self._image_resize)
self.bridge.register_method("image_generate_preview", self._image_generate_preview)
self.bridge.register_method("image_convert", self._image_convert)
self.bridge.register_method("notification_add", self.memory._add_notification)
self.bridge.register_method("notifications_get", self.memory._get_notifications)
self.bridge.register_method("notification_delete", self.memory._delete_notification)
self.bridge.register_method("notifications_expired_clean", self.memory._notifications_expired_clean)
await self.memory.initialise()

216
libervia/backend/memory/memory.py
View file

@ -16,29 +16,39 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os.path
import copy
import shortuuid
import mimetypes
import time
from functools import partial
from typing import Optional, Tuple, Dict
from pathlib import Path
from uuid import uuid4
from collections import namedtuple
import copy
from dataclasses import dataclass
from functools import partial
import mimetypes
import os.path
from pathlib import Path
import time
from typing import Dict, Optional, Tuple
from uuid import uuid4
import shortuuid
from twisted.internet import defer, error, reactor
from twisted.python import failure
from twisted.internet import defer, reactor, error
from twisted.words.protocols.jabber import jid
from libervia.backend.core.i18n import _
from libervia.backend.core.log import getLogger
from libervia.backend.core import exceptions
from libervia.backend.core.constants import Const as C
from libervia.backend.memory.sqla import Storage
from libervia.backend.memory.persistent import PersistentDict
from libervia.backend.memory.params import Params
from libervia.backend.memory.disco import Discovery
from libervia.backend.core.core_types import SatXMPPEntity
from libervia.backend.core.i18n import _
from libervia.backend.core.log import getLogger
from libervia.backend.memory.crypto import BlockCipher
from libervia.backend.memory.crypto import PasswordHasher
from libervia.backend.memory.disco import Discovery
from libervia.backend.memory.params import Params
from libervia.backend.memory.persistent import PersistentDict
from libervia.backend.memory.sqla import (
Notification,
NotificationPriority,
NotificationStatus,
NotificationType,
Storage,
)
from libervia.backend.tools import config as tools_config
from libervia.backend.tools.common import data_format
from libervia.backend.tools.common import regex
@ -1848,6 +1858,180 @@ class Memory:
*(regex.path_escape(a) for a in args)
)
## Notifications ##
def _add_notification(
self,
type_: str,
body_plain: str,
body_rich: str,
title: str,
is_global: bool,
requires_action: bool,
priority: str,
expire_at: float,
extra_s: str,
profile_key: str
) -> defer.Deferred:
client = self.host.get_client(profile_key)
if not client.is_admin:
raise exceptions.PermissionError("Only admins can add a notification")
try:
notification_type = NotificationType[type_]
notification_priority = NotificationPriority[priority]
except KeyError as e:
raise exceptions.DataError(
f"invalid notification type or priority data: {e}"
)
return defer.ensureDeferred(
self.add_notification(
client,
notification_type,
body_plain,
body_rich or None,
title or None,
is_global,
requires_action,
notification_priority,
expire_at or None,
data_format.deserialise(extra_s)
)
)
async def add_notification(
self,
client: SatXMPPEntity,
type_: NotificationType,
body_plain: str,
body_rich: Optional[str] = None,
title: Optional[str] = None,
is_global: bool = False,
requires_action: bool = False,
priority: NotificationPriority = NotificationPriority.MEDIUM,
expire_at: Optional[float] = None,
extra: Optional[dict] = None,
) -> None:
"""Create and broadcast a new notification.
@param client: client associated with the notification. If None, the notification
will be global (i.e. for all profiles).
@param type_: type of the notification.
@param body_plain: plain text body.
@param body_rich: rich text (XHTML) body.
@param title: optional title.
@param is_global: True if the notification is for all profiles.
@param requires_action: True if the notification requires user action (e.g. a
dialog need to be answered).
@priority: how urgent the notification is
@param expire_at: expiration timestamp for the notification.
@param extra: additional data.
"""
notification = await self.storage.add_notification(
None if is_global else client, type_, body_plain, body_rich, title,
requires_action, priority, expire_at, extra
)
self.host.bridge.notification_new(
str(notification.id),
notification.timestamp,
type_.value,
body_plain,
body_rich or '',
title or '',
requires_action,
priority.value,
expire_at or 0,
data_format.serialise(extra) if extra else '',
C.PROF_KEY_ALL if is_global else client.profile
)
def _get_notifications(self, filters_s: str, profile_key: str) -> defer.Deferred:
"""Fetch notifications for bridge with given filters and profile key.
@param filters_s: serialized filter conditions. Keys can be:
:type_ (str):
Filter by type of the notification.
:status (str):
Filter by status of the notification.
:requires_action (bool):
Filter by notifications that require user action.
:min_priority (str):
Filter by minimum priority value.
@param profile_key: key of the profile to fetch notifications for.
@return: Deferred which fires with a list of serialised notifications.
"""
client = self.host.get_client(profile_key)
filters = data_format.deserialise(filters_s)
try:
if 'type' in filters:
filters['type_'] = NotificationType[filters.pop('type')]
if 'status' in filters:
filters['status'] = NotificationStatus[filters['status']]
if 'min_priority' in filters:
filters['min_priority'] = NotificationPriority[filters['min_priority']].value
except KeyError as e:
raise exceptions.DataError(f"invalid filter data: {e}")
d = defer.ensureDeferred(self.storage.get_notifications(client, **filters))
d.addCallback(
lambda notifications: data_format.serialise(
[notification.serialise() for notification in notifications]
)
)
return d
def _delete_notification(
self,
id_: str,
is_global: bool,
profile_key: str
) -> defer.Deferred:
client = self.host.get_client(profile_key)
if is_global and not client.is_admin:
raise exceptions.PermissionError(
"Only admins can delete global notifications"
)
return defer.ensureDeferred(self.delete_notification(client, id_, is_global))
async def delete_notification(
self,
client: SatXMPPEntity,
id_: str,
is_global: bool=False
) -> None:
"""Delete a notification
the notification must be from the requesting profile.
@param id_: ID of the notification
is_global: if True, a global notification will be removed.
"""
await self.storage.delete_notification(None if is_global else client, id_)
self.host.bridge.notification_deleted(
id_,
C.PROF_KEY_ALL if is_global else client.profile
)
def _notifications_expired_clean(
self, limit_timestamp: float, profile_key: str
) -> defer.Deferred:
if profile_key == C.PROF_KEY_NONE:
client = None
else:
client = self.host.get_client(profile_key)
return defer.ensureDeferred(
self.storage.clean_expired_notifications(
client,
None if limit_timestamp == -1.0 else limit_timestamp
)
)
## Misc ##
def is_entity_available(self, client, entity_jid):

46
libervia/backend/memory/migration/versions/2ab01aa1f686_create_table_for_notifications.py Normal file
View file

@ -0,0 +1,46 @@
"""create table for notifications
Revision ID: 2ab01aa1f686
Revises: 4b002773cf92
Create Date: 2023-10-16 12:11:43.507295
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '2ab01aa1f686'
down_revision = '4b002773cf92'
branch_labels = None
depends_on = None
def upgrade():
op.create_table('notifications',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('timestamp', sa.Float(), nullable=False),
sa.Column('expire_at', sa.Float(), nullable=True),
sa.Column('profile_id', sa.Integer(), nullable=True),
sa.Column('type', sa.Enum('chat', 'blog', 'calendar', 'file', 'call', 'service', 'other', name='notificationtype'), nullable=False),
sa.Column('title', sa.Text(), nullable=True),
sa.Column('body_plain', sa.Text(), nullable=False),
sa.Column('body_rich', sa.Text(), nullable=True),
sa.Column('requires_action', sa.Boolean(), nullable=True),
sa.Column('priority', sa.Integer(), nullable=True),
sa.Column('extra_data', sa.JSON(), nullable=True),
sa.Column('status', sa.Enum('new', 'read', name='notificationstatus'), nullable=True),
sa.ForeignKeyConstraint(['profile_id'], ['profiles.id'], name=op.f('fk_notifications_profile_id_profiles'), ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id', name=op.f('pk_notifications'))
)
with op.batch_alter_table('notifications', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_notifications_profile_id'), ['profile_id'], unique=False)
batch_op.create_index('notifications_profile_id_status', ['profile_id', 'status'], unique=False)
def downgrade():
with op.batch_alter_table('notifications', schema=None) as batch_op:
batch_op.drop_index('notifications_profile_id_status')
batch_op.drop_index(batch_op.f('ix_notifications_profile_id'))
op.drop_table('notifications')

724
libervia/backend/memory/sqla.py
View file

File diff suppressed because it is too large Load diff

215
libervia/backend/memory/sqla_mapping.py
View file

@ -16,12 +16,12 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from typing import Dict, Any
from datetime import datetime
import enum
import json
import pickle
import time
from typing import Any, Dict
from sqlalchemy import (
Boolean,
@ -45,21 +45,52 @@ from sqlalchemy.types import TypeDecorator
from twisted.words.protocols.jabber import jid
from wokkel import generic
from libervia.backend.core.constants import Const as C
Base = declarative_base(
metadata=MetaData(
naming_convention={
"ix": 'ix_%(column_0_label)s',
"ix": "ix_%(column_0_label)s",
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s"
"pk": "pk_%(table_name)s",
}
)
)
# keys which are in message data extra but not stored in extra field this is
# because those values are stored in separate fields
NOT_IN_EXTRA = ('origin_id', 'stanza_id', 'received_timestamp', 'update_uid')
NOT_IN_EXTRA = ("origin_id", "stanza_id", "received_timestamp", "update_uid")
class Profiles(dict):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.id_to_profile = {v: k for k, v in self.items()}
def __setitem__(self, key, value):
super().__setitem__(key, value)
self.id_to_profile[value] = key
def __delitem__(self, key):
del self.id_to_profile[self[key]]
super().__delitem__(key)
def update(self, *args, **kwargs):
super().update(*args, **kwargs)
self.id_to_profile = {v: k for k, v in self.items()}
def clear(self):
super().clear()
self.id_to_profile.clear()
profiles = Profiles()
def get_profile_by_id( profile_id):
return profiles.id_to_profile.get(profile_id)
class SyncState(enum.Enum):
@ -78,11 +109,34 @@ class SubscriptionState(enum.Enum):
PENDING = 2
class NotificationType(enum.Enum):
chat = "chat"
blog = "blog"
calendar = "calendar"
file = "file"
call = "call"
service = "service"
other = "other"
class NotificationStatus(enum.Enum):
new = "new"
read = "read"
class NotificationPriority(enum.IntEnum):
LOW = 10
MEDIUM = 20
HIGH = 30
URGENT = 40
class LegacyPickle(TypeDecorator):
"""Handle troubles with data pickled by former version of SàT
This type is temporary until we do migration to a proper data type
"""
# Blob is used on SQLite but gives errors when used here, while Text works fine
impl = Text
cache_ok = True
@ -109,12 +163,13 @@ class LegacyPickle(TypeDecorator):
# JSON
return pickle.loads(
value.replace(b"sat.plugins", b"libervia.backend.plugins"),
encoding="utf-8"
encoding="utf-8",
)
class Json(TypeDecorator):
"""Handle JSON field in DB independant way"""
# Blob is used on SQLite but gives errors when used here, while Text works fine
impl = Text
cache_ok = True
@ -156,6 +211,7 @@ class Xml(TypeDecorator):
class JID(TypeDecorator):
"""Store twisted JID in text fields"""
impl = Text
cache_ok = True
@ -193,9 +249,7 @@ class Component(Base):
__tablename__ = "components"
profile_id = Column(
ForeignKey("profiles.id", ondelete="CASCADE"),
nullable=True,
primary_key=True
ForeignKey("profiles.id", ondelete="CASCADE"), nullable=True, primary_key=True
)
entry_point = Column(Text, nullable=False)
profile = relationship("Profile")
@ -209,7 +263,7 @@ class History(Base):
Index("history__profile_id_timestamp", "profile_id", "timestamp"),
Index(
"history__profile_id_received_timestamp", "profile_id", "received_timestamp"
)
),
)
uid = Column(Text, primary_key=True)
@ -295,14 +349,14 @@ class History(Base):
if self.thread.parent_id is not None:
extra["thread_parent"] = self.thread.parent_id
return {
"from": f"{self.source}/{self.source_res}" if self.source_res
else self.source,
"from": f"{self.source}/{self.source_res}"
if self.source_res
else self.source,
"to": f"{self.dest}/{self.dest_res}" if self.dest_res else self.dest,
"uid": self.uid,
"message": {m.language or '': m.message for m in self.messages},
"subject": {m.language or '': m.subject for m in self.subjects},
"message": {m.language or "": m.message for m in self.messages},
"subject": {m.language or "": m.subject for m in self.subjects},
"type": self.type,
"extra": extra,
"timestamp": self.timestamp,
@ -311,8 +365,14 @@ class History(Base):
def as_tuple(self):
d = self.serialise()
return (
d['uid'], d['timestamp'], d['from'], d['to'], d['message'], d['subject'],
d['type'], d['extra']
d["uid"],
d["timestamp"],
d["from"],
d["to"],
d["message"],
d["subject"],
d["type"],
d["extra"],
)
@staticmethod
@ -336,9 +396,7 @@ class History(Base):
class Message(Base):
__tablename__ = "message"
__table_args__ = (
Index("message__history_uid", "history_uid"),
)
__table_args__ = (Index("message__history_uid", "history_uid"),)
id = Column(
Integer,
@ -358,16 +416,14 @@ class Message(Base):
def __repr__(self):
lang_str = f"[{self.language}]" if self.language else ""
msg = f"{self.message[:20]}…" if len(self.message)>20 else self.message
msg = f"{self.message[:20]}…" if len(self.message) > 20 else self.message
content = f"{lang_str}{msg}"
return f"Message<{content}>"
class Subject(Base):
__tablename__ = "subject"
__table_args__ = (
Index("subject__history_uid", "history_uid"),
)
__table_args__ = (Index("subject__history_uid", "history_uid"),)
id = Column(
Integer,
@ -387,16 +443,14 @@ class Subject(Base):
def __repr__(self):
lang_str = f"[{self.language}]" if self.language else ""
msg = f"{self.subject[:20]}…" if len(self.subject)>20 else self.subject
msg = f"{self.subject[:20]}…" if len(self.subject) > 20 else self.subject
content = f"{lang_str}{msg}"
return f"Subject<{content}>"
class Thread(Base):
__tablename__ = "thread"
__table_args__ = (
Index("thread__history_uid", "history_uid"),
)
__table_args__ = (Index("thread__history_uid", "history_uid"),)
id = Column(
Integer,
@ -412,6 +466,51 @@ class Thread(Base):
return f"Thread<{self.thread_id} [parent: {self.parent_id}]>"
class Notification(Base):
__tablename__ = "notifications"
__table_args__ = (Index("notifications_profile_id_status", "profile_id", "status"),)
id = Column(Integer, primary_key=True, autoincrement=True)
timestamp = Column(Float, nullable=False, default=time.time)
expire_at = Column(Float, nullable=True)
profile_id = Column(ForeignKey("profiles.id", ondelete="CASCADE"), index=True, nullable=True)
profile = relationship("Profile")
type = Column(Enum(NotificationType), nullable=False)
title = Column(Text, nullable=True)
body_plain = Column(Text, nullable=False)
body_rich = Column(Text, nullable=True)
requires_action = Column(Boolean, default=False)
priority = Column(Integer, default=NotificationPriority.MEDIUM.value)
extra_data = Column(JSON)
status = Column(Enum(NotificationStatus), default=NotificationStatus.new)
def serialise(self) -> dict[str, str | float | bool | int | dict]:
"""
Serialises the Notification instance to a dictionary.
"""
result = {}
for column in self.__table__.columns:
value = getattr(self, column.name)
if value is not None:
if column.name in ("type", "status"):
result[column.name] = value.name
elif column.name == "id":
result[column.name] = str(value)
elif column.name == "profile_id":
if value is None:
result["profile"] = C.PROF_KEY_ALL
else:
result["profile"] = get_profile_by_id(value)
else:
result[column.name] = value
return result
class ParamGen(Base):
__tablename__ = "param_gen"
@ -425,9 +524,7 @@ class ParamInd(Base):
category = Column(Text, primary_key=True)
name = Column(Text, primary_key=True)
profile_id = Column(
ForeignKey("profiles.id", ondelete="CASCADE"), primary_key=True
)
profile_id = Column(ForeignKey("profiles.id", ondelete="CASCADE"), primary_key=True)
value = Column(Text)
profile = relationship("Profile", back_populates="params")
@ -446,9 +543,7 @@ class PrivateInd(Base):
namespace = Column(Text, primary_key=True)
key = Column(Text, primary_key=True)
profile_id = Column(
ForeignKey("profiles.id", ondelete="CASCADE"), primary_key=True
)
profile_id = Column(ForeignKey("profiles.id", ondelete="CASCADE"), primary_key=True)
value = Column(Text)
profile = relationship("Profile", back_populates="private_data")
@ -467,9 +562,7 @@ class PrivateIndBin(Base):
namespace = Column(Text, primary_key=True)
key = Column(Text, primary_key=True)
profile_id = Column(
ForeignKey("profiles.id", ondelete="CASCADE"), primary_key=True
)
profile_id = Column(ForeignKey("profiles.id", ondelete="CASCADE"), primary_key=True)
value = Column(LegacyPickle)
profile = relationship("Profile", back_populates="private_bin_data")
@ -484,8 +577,8 @@ class File(Base):
"profile_id",
"owner",
"media_type",
"media_subtype"
)
"media_subtype",
),
)
id = Column(Text, primary_key=True)
@ -493,11 +586,7 @@ class File(Base):
version = Column(Text, primary_key=True)
parent = Column(Text, nullable=False)
type = Column(
Enum(
"file", "directory",
name="file_type",
create_constraint=True
),
Enum("file", "directory", name="file_type", create_constraint=True),
nullable=False,
server_default="file",
)
@ -520,14 +609,10 @@ class File(Base):
class PubsubNode(Base):
__tablename__ = "pubsub_nodes"
__table_args__ = (
UniqueConstraint("profile_id", "service", "name"),
)
__table_args__ = (UniqueConstraint("profile_id", "service", "name"),)
id = Column(Integer, primary_key=True)
profile_id = Column(
ForeignKey("profiles.id", ondelete="CASCADE")
)
profile_id = Column(ForeignKey("profiles.id", ondelete="CASCADE"))
service = Column(JID)
name = Column(Text, nullable=False)
subscribed = Column(
@ -540,19 +625,11 @@ class PubsubNode(Base):
name="sync_state",
create_constraint=True,
),
nullable=True
)
sync_state_updated = Column(
Float,
nullable=False,
default=time.time()
)
type_ = Column(
Text, name="type", nullable=True
)
subtype = Column(
Text, nullable=True
nullable=True,
)
sync_state_updated = Column(Float, nullable=False, default=time.time())
type_ = Column(Text, name="type", nullable=True)
subtype = Column(Text, nullable=True)
extra = Column(JSON)
items = relationship("PubsubItem", back_populates="node", passive_deletes=True)
@ -567,10 +644,9 @@ class PubsubSub(Base):
Used by components managing a pubsub service
"""
__tablename__ = "pubsub_subs"
__table_args__ = (
UniqueConstraint("node_id", "subscriber"),
)
__table_args__ = (UniqueConstraint("node_id", "subscriber"),)
id = Column(Integer, primary_key=True)
node_id = Column(ForeignKey("pubsub_nodes.id", ondelete="CASCADE"), nullable=False)
@ -581,7 +657,7 @@ class PubsubSub(Base):
name="state",
create_constraint=True,
),
nullable=True
nullable=True,
)
node = relationship("PubsubNode", back_populates="subscriptions")
@ -589,9 +665,7 @@ class PubsubSub(Base):
class PubsubItem(Base):
__tablename__ = "pubsub_items"
__table_args__ = (
UniqueConstraint("node_id", "name"),
)
__table_args__ = (UniqueConstraint("node_id", "name"),)
id = Column(Integer, primary_key=True)
node_id = Column(ForeignKey("pubsub_nodes.id", ondelete="CASCADE"), nullable=False)
name = Column(Text, nullable=False)
@ -607,6 +681,7 @@ class PubsubItem(Base):
# create
@event.listens_for(PubsubItem.__table__, "after_create")
def fts_create(target, connection, **kw):
"""Full-Text Search table creation"""
@ -626,13 +701,15 @@ def fts_create(target, connection, **kw):
" INSERT INTO pubsub_items_fts(pubsub_items_fts, rowid, data) VALUES"
"('delete', old.id, old.data);"
" INSERT INTO pubsub_items_fts(rowid, data) VALUES(new.id, new.data);"
"END"
"END",
]
for q in queries:
connection.execute(DDL(q))
# drop
@event.listens_for(PubsubItem.__table__, "before_drop")
def fts_drop(target, connection, **kw):
"Full-Text Search table drop" ""

88
libervia/frontends/bridge/dbus_bridge.py
View file

@ -562,6 +562,62 @@ class bridge:
kwargs['error_handler'] = error_handler
return self.db_core_iface.namespaces_get(**kwargs)
def notification_add(self, type_, body_plain, body_rich, title, is_global, requires_action, arg_6, priority, expire_at, extra, callback=None, errback=None):
if callback is None:
error_handler = None
else:
if errback is None:
errback = log.error
error_handler = lambda err:errback(dbus_to_bridge_exception(err))
kwargs={}
if callback is not None:
kwargs['timeout'] = const_TIMEOUT
kwargs['reply_handler'] = callback
kwargs['error_handler'] = error_handler
return self.db_core_iface.notification_add(type_, body_plain, body_rich, title, is_global, requires_action, arg_6, priority, expire_at, extra, **kwargs)
def notification_delete(self, id_, is_global, profile_key, callback=None, errback=None):
if callback is None:
error_handler = None
else:
if errback is None:
errback = log.error
error_handler = lambda err:errback(dbus_to_bridge_exception(err))
kwargs={}
if callback is not None:
kwargs['timeout'] = const_TIMEOUT
kwargs['reply_handler'] = callback
kwargs['error_handler'] = error_handler
return self.db_core_iface.notification_delete(id_, is_global, profile_key, **kwargs)
def notifications_expired_clean(self, limit_timestamp, profile_key, callback=None, errback=None):
if callback is None:
error_handler = None
else:
if errback is None:
errback = log.error
error_handler = lambda err:errback(dbus_to_bridge_exception(err))
kwargs={}
if callback is not None:
kwargs['timeout'] = const_TIMEOUT
kwargs['reply_handler'] = callback
kwargs['error_handler'] = error_handler
return self.db_core_iface.notifications_expired_clean(limit_timestamp, profile_key, **kwargs)
def notifications_get(self, filters, profile_key, callback=None, errback=None):
if callback is None:
error_handler = None
else:
if errback is None:
errback = log.error
error_handler = lambda err:errback(dbus_to_bridge_exception(err))
kwargs={}
if callback is not None:
kwargs['timeout'] = const_TIMEOUT
kwargs['reply_handler'] = callback
kwargs['error_handler'] = error_handler
return str(self.db_core_iface.notifications_get(filters, profile_key, **kwargs))
def param_get_a(self, name, category, attribute="value", profile_key="@DEFAULT@", callback=None, errback=None):
if callback is None:
error_handler = None
@ -1271,6 +1327,38 @@ class AIOBridge(bridge):
self.db_core_iface.namespaces_get(timeout=const_TIMEOUT, reply_handler=reply_handler, error_handler=error_handler)
return fut
def notification_add(self, type_, body_plain, body_rich, title, is_global, requires_action, arg_6, priority, expire_at, extra):
loop = asyncio.get_running_loop()
fut = loop.create_future()
reply_handler = lambda ret=None: loop.call_soon_threadsafe(fut.set_result, ret)
error_handler = lambda err: loop.call_soon_threadsafe(fut.set_exception, dbus_to_bridge_exception(err))
self.db_core_iface.notification_add(type_, body_plain, body_rich, title, is_global, requires_action, arg_6, priority, expire_at, extra, timeout=const_TIMEOUT, reply_handler=reply_handler, error_handler=error_handler)
return fut
def notification_delete(self, id_, is_global, profile_key):
loop = asyncio.get_running_loop()
fut = loop.create_future()
reply_handler = lambda ret=None: loop.call_soon_threadsafe(fut.set_result, ret)
error_handler = lambda err: loop.call_soon_threadsafe(fut.set_exception, dbus_to_bridge_exception(err))
self.db_core_iface.notification_delete(id_, is_global, profile_key, timeout=const_TIMEOUT, reply_handler=reply_handler, error_handler=error_handler)
return fut
def notifications_expired_clean(self, limit_timestamp, profile_key):
loop = asyncio.get_running_loop()
fut = loop.create_future()
reply_handler = lambda ret=None: loop.call_soon_threadsafe(fut.set_result, ret)
error_handler = lambda err: loop.call_soon_threadsafe(fut.set_exception, dbus_to_bridge_exception(err))
self.db_core_iface.notifications_expired_clean(limit_timestamp, profile_key, timeout=const_TIMEOUT, reply_handler=reply_handler, error_handler=error_handler)
return fut
def notifications_get(self, filters, profile_key):
loop = asyncio.get_running_loop()
fut = loop.create_future()
reply_handler = lambda ret=None: loop.call_soon_threadsafe(fut.set_result, ret)
error_handler = lambda err: loop.call_soon_threadsafe(fut.set_exception, dbus_to_bridge_exception(err))
self.db_core_iface.notifications_get(filters, profile_key, timeout=const_TIMEOUT, reply_handler=reply_handler, error_handler=error_handler)
return fut
def param_get_a(self, name, category, attribute="value", profile_key="@DEFAULT@"):
loop = asyncio.get_running_loop()
fut = loop.create_future()

56
libervia/frontends/bridge/pb.py
View file

@ -489,6 +489,42 @@ class bridge(object):
else:
d.addErrback(self._errback, ori_errback=errback)
def notification_add(self, type_, body_plain, body_rich, title, is_global, requires_action, arg_6, priority, expire_at, extra, callback=None, errback=None):
d = self.root.callRemote("notification_add", type_, body_plain, body_rich, title, is_global, requires_action, arg_6, priority, expire_at, extra)
if callback is not None:
d.addCallback(lambda __: callback())
if errback is None:
d.addErrback(self._generic_errback)
else:
d.addErrback(self._errback, ori_errback=errback)
def notification_delete(self, id_, is_global, profile_key, callback=None, errback=None):
d = self.root.callRemote("notification_delete", id_, is_global, profile_key)
if callback is not None:
d.addCallback(lambda __: callback())
if errback is None:
d.addErrback(self._generic_errback)
else:
d.addErrback(self._errback, ori_errback=errback)
def notifications_expired_clean(self, limit_timestamp, profile_key, callback=None, errback=None):
d = self.root.callRemote("notifications_expired_clean", limit_timestamp, profile_key)
if callback is not None:
d.addCallback(lambda __: callback())
if errback is None:
d.addErrback(self._generic_errback)
else:
d.addErrback(self._errback, ori_errback=errback)
def notifications_get(self, filters, profile_key, callback=None, errback=None):
d = self.root.callRemote("notifications_get", filters, profile_key)
if callback is not None:
d.addCallback(callback)
if errback is None:
d.addErrback(self._generic_errback)
else:
d.addErrback(self._errback, ori_errback=errback)
def param_get_a(self, name, category, attribute="value", profile_key="@DEFAULT@", callback=None, errback=None):
d = self.root.callRemote("param_get_a", name, category, attribute, profile_key)
if callback is not None:
@ -969,6 +1005,26 @@ class AIOBridge(bridge):
d.addErrback(self._errback)
return d.asFuture(asyncio.get_event_loop())
def notification_add(self, type_, body_plain, body_rich, title, is_global, requires_action, arg_6, priority, expire_at, extra):
d = self.root.callRemote("notification_add", type_, body_plain, body_rich, title, is_global, requires_action, arg_6, priority, expire_at, extra)
d.addErrback(self._errback)
return d.asFuture(asyncio.get_event_loop())
def notification_delete(self, id_, is_global, profile_key):
d = self.root.callRemote("notification_delete", id_, is_global, profile_key)
d.addErrback(self._errback)
return d.asFuture(asyncio.get_event_loop())
def notifications_expired_clean(self, limit_timestamp, profile_key):
d = self.root.callRemote("notifications_expired_clean", limit_timestamp, profile_key)
d.addErrback(self._errback)
return d.asFuture(asyncio.get_event_loop())
def notifications_get(self, filters, profile_key):
d = self.root.callRemote("notifications_get", filters, profile_key)
d.addErrback(self._errback)
return d.asFuture(asyncio.get_event_loop())
def param_get_a(self, name, category, attribute="value", profile_key="@DEFAULT@"):
d = self.root.callRemote("param_get_a", name, category, attribute, profile_key)
d.addErrback(self._errback)