erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
libervia-backend/libervia/backend/memory/sqla.py
Goffi 50ffc8c3b4 core, memory (sqla), plugins XEP-0277, XEP-0346: various fixes: 
- `is_local`: now use `server_jid`.
- `aio_func`: pass the missing `self` argument (`aio_fund` is unbound).
- Declare `pubsub_service` in `core_types.SatXMPPEntity`
- plugin XEP-0277: fix handling of `MbData.from_element` when node is not specified (so we
  use default node).
- plugin XEP-0346: `IdentityData`: fix field access, following move to Pydantic model.
2026-10-27 16:04:50 +01:00

2808 lines
107 KiB
Python
Raw Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
# Libervia: an XMPP client
# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org)
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# 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 asyncio
from asyncio.subprocess import PIPE
from collections.abc import Callable, Iterable
from datetime import datetime
import json
from pathlib import Path
import sys
import time
from typing import Any, Dict, List, Literal, Optional, Tuple, Union
from alembic import config as al_config, script as al_script
from alembic.runtime import migration as al_migration
from sqlalchemy import (
and_,
case,
delete,
desc,
event,
exists,
func,
literal,
or_,
over,
true,
update,
)
from sqlalchemy import Integer, literal_column, text
from sqlalchemy.dialects.sqlite import insert
from sqlalchemy.engine import Connection, Engine
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
from sqlalchemy.future import select
from sqlalchemy.orm import (
contains_eager,
joinedload,
selectinload,
sessionmaker,
subqueryload,
)
from sqlalchemy.orm.attributes import Mapped
from sqlalchemy.orm.decl_api import DeclarativeMeta
from sqlalchemy.sql.functions import coalesce, count, now, sum as sum_
from twisted.internet import defer
from twisted.words.protocols.jabber import jid
from twisted.words.xish import domish
from libervia.backend.core import exceptions
from libervia.backend.core.constants import Const as C
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 import migration
from libervia.backend.memory import sqla_config
from libervia.backend.memory.sqla_mapping import (
AccessModel,
Affiliation,
Base,
Component,
File,
History,
Message,
NOT_IN_EXTRA,
Notification,
NotificationPriority,
NotificationStatus,
NotificationType,
ParamGen,
ParamInd,
PrivateGen,
PrivateGenBin,
PrivateInd,
PrivateIndBin,
Profile,
PublishModel,
PubsubAffiliation,
PubsubItem,
PubsubNode,
PubsubSub,
Subject,
SubscriptionState,
SyncState,
Thread,
get_profile_by_id,
profiles,
)
from libervia.backend.tools.common import uri
from libervia.backend.tools.utils import aio, as_future
log = getLogger(__name__)
migration_path = Path(migration.__file__).parent
#: mapping of Libervia search query operators to SQLAlchemy method name
OP_MAP = {
"==": "__eq__",
"eq": "__eq__",
"!=": "__ne__",
"ne": "__ne__",
">": "__gt__",
"gt": "__gt__",
"<": "__le__",
"le": "__le__",
"between": "between",
"in": "in_",
"not_in": "not_in",
"overlap": "in_",
"ioverlap": "in_",
"disjoint": "in_",
"idisjoint": "in_",
"like": "like",
"ilike": "ilike",
"not_like": "notlike",
"not_ilike": "notilike",
}
@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
class Storage:
def __init__(self):
self.initialized = defer.Deferred()
self.engine: AsyncEngine
# we keep cache for the profiles (key: profile name, value: profile id)
# profile id to component entry point
self.components: Dict[int, str] = {}
@property
def profiles(self):
return profiles
def get_profile_by_id(self, profile_id):
return get_profile_by_id(profile_id)
def get_profile_by_name(self, profile: str) -> str:
return self.profiles[profile]
@aio
async def unload(self) -> None:
"""Clean up database resources.
Flushes WAL and disposes engine.
This method is called on shutdown.
"""
if self.engine.dialect.name == "sqlite":
await self.wal_checkpoint.aio_func(self)
await self.engine.dispose()
async def migrate_apply(self, *args: str, log_output: bool = False) -> None:
"""Do a migration command
Commands are applied by running Alembic in a subprocess.
Arguments are alembic executables commands
@param log_output: manage stdout and stderr:
- if False, stdout and stderr are buffered, and logged only in case of error
- if True, stdout and stderr will be logged during the command execution
@raise exceptions.DatabaseError: something went wrong while running the
process
"""
stdout, stderr = 2 * (None,) if log_output else 2 * (PIPE,)
proc = await asyncio.create_subprocess_exec(
sys.executable,
"-m",
"alembic",
*args,
stdout=stdout,
stderr=stderr,
cwd=migration_path,
)
log_out, log_err = await proc.communicate()
if proc.returncode != 0:
msg = _("Can't {operation} database (exit code {exit_code})").format(
operation=args[0], exit_code=proc.returncode
)
if log_out or log_err:
msg += f":\nstdout: {log_out.decode()}\nstderr: {log_err.decode()}"
log.error(msg)
raise exceptions.DatabaseError(msg)
async def create_db(self, engine: AsyncEngine, db_config: dict) -> None:
"""Create a new database
The database is generated from SQLAlchemy model, then stamped by Alembic
"""
# the dir may not exist if it's not the XDG recommended one
db_config["path"].parent.mkdir(0o700, True, True)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
log.debug("stamping the database")
await self.migrate_apply("stamp", "head")
log.debug("stamping done")
def _check_db_is_up_to_date(self, conn: Connection) -> bool:
al_ini_path = migration_path / "alembic.ini"
al_cfg = al_config.Config(al_ini_path)
directory = al_script.ScriptDirectory.from_config(al_cfg)
context = al_migration.MigrationContext.configure(conn)
return set(context.get_current_heads()) == set(directory.get_heads())
def _sqlite_set_journal_mode_wal(self, conn: Connection) -> None:
"""Check if journal mode is WAL, set it if necessary and checkpoint
If WAL was already active from a previous run, checkpoints at startup
so the main DB has all data, avoiding slow reads on first login.
"""
result = conn.execute(text("PRAGMA journal_mode"))
if result.scalar() != "wal":
log.info("WAL mode not activated, activating it")
conn.execute(text("PRAGMA journal_mode=WAL"))
else:
# WAL was already active, checkpoint at startup
result = conn.execute(text("PRAGMA wal_checkpoint(TRUNCATE)"))
log.debug(f"WAL checkpoint at startup: {result.fetchone()}")
async def check_and_update_db(self, engine: AsyncEngine, db_config: dict) -> None:
"""Check that database is up-to-date, and update if necessary"""
async with engine.connect() as conn:
up_to_date = await conn.run_sync(self._check_db_is_up_to_date)
if up_to_date:
log.debug("Database is up-to-date")
else:
log.info("Database needs to be updated")
log.info("updating…")
await self.migrate_apply("upgrade", "head", log_output=True)
log.info("Database is now up-to-date")
@aio
async def initialise(self) -> None:
log.info(_("Connecting database"))
db_config = sqla_config.get_db_config()
engine = self.engine = create_async_engine(
db_config["url"],
future=True,
json_serializer=lambda obj: json.dumps(obj, ensure_ascii=False),
)
new_base = not db_config["path"].exists()
if new_base:
log.info(_("The database is new, creating the tables"))
await self.create_db(engine, db_config)
else:
await self.check_and_update_db(engine, db_config)
async with engine.connect() as conn:
await conn.run_sync(self._sqlite_set_journal_mode_wal)
self.session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
async with self.session() as session:
result = await session.execute(select(Profile))
for p in result.scalars():
self.profiles[p.name] = p.id
result = await session.execute(select(Component))
for c in result.scalars():
self.components[c.profile_id] = c.entry_point
self.initialized.callback(None)
@aio
async def wal_checkpoint(
self, mode: Literal["PASSIVE", "FULL", "RESTART", "TRUNCATE", "NOOP"] = "TRUNCATE"
) -> tuple[int, int, int]:
"""Checkpoint WAL file to flush pending changes to main DB
See https://sqlite.org/pragma.html#pragma_wal_checkpoint for details.
@param mode: checkpoint mode
@return: (num_checkpointed, num_dirty_pages, num_pages)
"""
async with self.session() as session:
result = await session.execute(text(f"PRAGMA wal_checkpoint({mode})"))
fetched = result.fetchone()
log.debug(
f"WAL checkpoint ({mode}): "
f"checkpointed={fetched[0]}, dirty_pages={fetched[1]}, pages={fetched[2]}"
)
return fetched
## Generic
@aio
async def get(
self,
client: SatXMPPEntity,
db_cls: DeclarativeMeta,
db_id_col: Mapped,
id_value: Any,
joined_loads=None,
) -> Optional[DeclarativeMeta]:
stmt = select(db_cls).where(db_id_col == id_value)
if client is not None:
stmt = stmt.filter_by(profile_id=self.profiles[client.profile])
if joined_loads is not None:
for joined_load in joined_loads:
stmt = stmt.options(joinedload(joined_load))
async with self.session() as session:
result = await session.execute(stmt)
if joined_loads is not None:
result = result.unique()
return result.scalar_one_or_none()
@aio
async def add(self, db_obj: DeclarativeMeta) -> None:
"""Add an object to database"""
async with self.session() as session:
async with session.begin():
session.add(db_obj)
@aio
async def delete(
self,
db_obj: Union[DeclarativeMeta, List[DeclarativeMeta]],
session_add: Optional[List[DeclarativeMeta]] = None,
) -> None:
"""Delete an object from database
@param db_obj: object to delete or list of objects to delete
@param session_add: other objects to add to session.
This is useful when parents of deleted objects needs to be updated too, or if
other objects needs to be updated in the same transaction.
"""
if not db_obj:
return
if not isinstance(db_obj, list):
db_obj = [db_obj]
async with self.session() as session:
async with session.begin():
if session_add is not None:
for obj in session_add:
session.add(obj)
for obj in db_obj:
await session.delete(obj)
await session.commit()
## Profiles
def get_profiles_list(self) -> List[str]:
""" "Return list of all registered profiles"""
return list(self.profiles.keys())
def has_profile(self, profile_name: str) -> bool:
"""return True if profile_name exists
@param profile_name: name of the profile to check
"""
return profile_name in self.profiles
def profile_is_component(self, profile_name: str) -> bool:
try:
return self.profiles[profile_name] in self.components
except KeyError:
raise exceptions.NotFound("the requested profile doesn't exists")
def get_entry_point(self, profile_name: str) -> str:
try:
return self.components[self.profiles[profile_name]]
except KeyError:
raise exceptions.NotFound(
"the requested profile doesn't exists or is not a component"
)
@aio
async def create_profile(self, name: str, component_ep: str | None = None) -> Profile:
"""Create a new profile
@param name: name of the profile
@param component_ep: if not None, must point to a component entry point
@return: the created profile
"""
async with self.session() as session:
profile = Profile(name=name)
async with session.begin():
session.add(profile)
self.profiles[profile.name] = profile.id
if component_ep is not None:
async with session.begin():
component = Component(profile=profile, entry_point=component_ep)
session.add(component)
self.components[profile.id] = component_ep
return profile
@aio
async def delete_profile(self, name: str) -> None:
"""Delete profile
@param name: name of the profile
"""
async with self.session() as session:
result = await session.execute(select(Profile).where(Profile.name == name))
profile = result.scalar()
await session.delete(profile)
await session.commit()
del self.profiles[profile.name]
if profile.id in self.components:
del self.components[profile.id]
log.info(_("Profile {name!r} deleted").format(name=name))
## Params
@aio
async def load_gen_params(self, params_gen: dict) -> None:
"""Load general parameters
@param params_gen: dictionary to fill
"""
log.debug(_("loading general parameters from database"))
async with self.session() as session:
result = await session.execute(select(ParamGen))
for p in result.scalars():
params_gen[(p.category, p.name)] = p.value
@aio
async def load_ind_params(self, params_ind: dict, profile: str) -> None:
"""Load individual parameters
@param params_ind: dictionary to fill
@param profile: a profile which *must* exist
"""
log.debug(_("loading individual parameters from database"))
async with self.session() as session:
result = await session.execute(
select(ParamInd).where(ParamInd.profile_id == self.profiles[profile])
)
for p in result.scalars():
params_ind[(p.category, p.name)] = p.value
@aio
async def get_ind_param(
self, category: str, name: str, profile: str
) -> Optional[str]:
"""Ask database for the value of one specific individual parameter
@param category: category of the parameter
@param name: name of the parameter
@param profile: %(doc_profile)s
"""
async with self.session() as session:
result = await session.execute(
select(ParamInd.value).filter_by(
category=category, name=name, profile_id=self.profiles[profile]
)
)
return result.scalar_one_or_none()
@aio
async def get_ind_param_values(self, category: str, name: str) -> Dict[str, str]:
"""Ask database for the individual values of a parameter for all profiles
@param category: category of the parameter
@param name: name of the parameter
@return dict: profile => value map
"""
async with self.session() as session:
result = await session.execute(
select(ParamInd)
.filter_by(category=category, name=name)
.options(subqueryload(ParamInd.profile))
)
return {param.profile.name: param.value for param in result.scalars()}
@aio
async def set_gen_param(self, category: str, name: str, value: Optional[str]) -> None:
"""Save the general parameters in database
@param category: category of the parameter
@param name: name of the parameter
@param value: value to set
"""
async with self.session() as session:
stmt = (
insert(ParamGen)
.values(category=category, name=name, value=value)
.on_conflict_do_update(
index_elements=(ParamGen.category, ParamGen.name),
set_={ParamGen.value: value},
)
)
await session.execute(stmt)
await session.commit()
@aio
async def set_ind_param(
self, category: str, name: str, value: Optional[str], profile: str
) -> None:
"""Save the individual parameters in database
@param category: category of the parameter
@param name: name of the parameter
@param value: value to set
@param profile: a profile which *must* exist
"""
async with self.session() as session:
stmt = (
insert(ParamInd)
.values(
category=category,
name=name,
profile_id=self.profiles[profile],
value=value,
)
.on_conflict_do_update(
index_elements=(
ParamInd.category,
ParamInd.name,
ParamInd.profile_id,
),
set_={ParamInd.value: value},
)
)
await session.execute(stmt)
await session.commit()
def _jid_filter(self, jid_: jid.JID, dest: bool = False):
"""Generate condition to filter on a JID, using relevant columns
@param dest: True if it's the destinee JID, otherwise it's the source one
@param jid_: JID to filter by
"""
if jid_.resource:
if dest:
return and_(
History.dest == jid_.userhost(), History.dest_res == jid_.resource
)
else:
return and_(
History.source == jid_.userhost(), History.source_res == jid_.resource
)
else:
if dest:
return History.dest == jid_.userhost()
else:
return History.source == jid_.userhost()
@aio
async def get_history_from_xmpp_id(
self, client: SatXMPPEntity, message_id: str, message_type: str | None
) -> History | None:
"""Retrieves history row that match a specific message ID.
The retrieval criteria vary based on the message type.
@param message_type: The type of the message (one of C.MESS_TYPE_*)
@param message_id: the ID of the message, XEP-0359's stanza-id is used for
groupchat, origin-id from the same XEP otherwise.
@return: History instance with messages, subjects and threads, if any found, None
otherwise.
"""
profile_id = self.profiles[client.profile]
async with self.session() as session:
if message_type == C.MESS_TYPE_GROUPCHAT:
query = select(History).where(
History.profile_id == profile_id, History.stanza_id == message_id
)
else:
query = select(History).where(
History.profile_id == profile_id, History.origin_id == message_id
)
query = (
query.outerjoin(History.messages)
.outerjoin(History.subjects)
.outerjoin(History.thread)
.options(
contains_eager(History.messages),
contains_eager(History.subjects),
contains_eager(History.thread),
)
)
result = await session.execute(query)
history = result.scalars().first()
return history
@aio
async def history_get(
self,
from_jid: jid.JID | None,
to_jid: jid.JID | None,
limit: int | None = None,
between: bool = True,
filters: dict[str, str | bool] | None = None,
profile: str | None = None,
) -> list[
tuple[str, float, str, str, dict[str, str], dict[str, str], str, dict[str, Any]]
]:
"""Retrieve messages in history
@param from_jid: source JID (full, or bare for catchall)
@param to_jid: dest JID (full, or bare for catchall)
@param limit: maximum number of messages to get:
- 0 for no message (returns the empty list)
- None for unlimited
@param between: confound source and dest (ignore the direction)
@param filters: pattern to filter the history results
@return: list of messages as in [message_new], minus the profile which is already
known.
"""
# we have to set a default value to profile because it's last argument
# and thus follow other keyword arguments with default values
# but None should not be used for it
assert profile is not None
if limit == 0:
return []
if filters is None:
filters = {}
# An "after_uid" filter asks for the first messages after a given one, so ASC
# order is used and the result is returned as-is. Otherwise DESC+LIMIT keeps the
# last messages and the result is reversed.
asc = "after_uid" in filters
order = (
History.timestamp.asc() if asc else History.timestamp.desc(),
(
History.received_timestamp.asc()
if asc
else History.received_timestamp.desc()
),
)
stmt = (
select(History)
.filter_by(profile_id=self.profiles[profile])
.outerjoin(History.messages)
.outerjoin(History.subjects)
.outerjoin(History.thread)
.options(
contains_eager(History.messages),
contains_eager(History.subjects),
contains_eager(History.thread),
)
.order_by(*order)
)
if not from_jid and not to_jid:
# no jid specified, we want all one2one communications
pass
elif between:
if not from_jid or not to_jid:
# we only have one jid specified, we check all messages
# from or to this jid
jid_ = from_jid or to_jid
stmt = stmt.where(
or_(self._jid_filter(jid_), self._jid_filter(jid_, dest=True))
)
else:
# we have 2 jids specified, we check all communications between
# those 2 jids
stmt = stmt.where(
or_(
and_(
self._jid_filter(from_jid),
self._jid_filter(to_jid, dest=True),
),
and_(
self._jid_filter(to_jid),
self._jid_filter(from_jid, dest=True),
),
)
)
else:
# we want one communication in specific direction (from somebody or
# to somebody).
if from_jid is not None:
stmt = stmt.where(self._jid_filter(from_jid))
if to_jid is not None:
stmt = stmt.where(self._jid_filter(to_jid, dest=True))
if filters:
if "timestamp_start" in filters:
stmt = stmt.where(History.timestamp >= float(filters["timestamp_start"]))
if "before_uid" in filters:
# orignially this query was using SQLITE's rowid. This has been changed to
# use coalesce(received_timestamp, timestamp) to be SQL engine independant
stmt = stmt.where(
coalesce(History.received_timestamp, History.timestamp)
< (
select(
coalesce(History.received_timestamp, History.timestamp)
).filter_by(uid=filters["before_uid"])
).scalar_subquery()
)
if "after_uid" in filters:
stmt = stmt.where(
coalesce(History.received_timestamp, History.timestamp)
> (
select(
coalesce(History.received_timestamp, History.timestamp)
).filter_by(uid=filters["after_uid"])
).scalar_subquery()
)
if "body" in filters:
# TODO: use REGEXP (function to be defined) instead of GLOB:
# https://www.sqlite.org/lang_expr.html
stmt = stmt.where(Message.message.like(f"%{filters['body']}%"))
if "search" in filters:
search_term = f"%{filters['search']}%"
stmt = stmt.where(
or_(
Message.message.like(search_term),
History.source_res.like(search_term),
)
)
if "types" in filters:
types = filters["types"].split()
stmt = stmt.where(History.type.in_(types))
if "not_types" in filters:
types = filters["not_types"].split()
stmt = stmt.where(History.type.not_in(types))
if "last_stanza_id" in filters:
# this request get the last message with a "stanza_id" that we
# have in history. This is mainly used to retrieve messages sent
# while we were offline, using MAM (XEP-0313).
if filters["last_stanza_id"] is not True or limit != 1:
raise ValueError("Unexpected values for last_stanza_id filter")
stmt = stmt.where(History.stanza_id.is_not(None))
if "id" in filters:
stmt = stmt.where(History.uid == filters["id"])
if "origin_id" in filters:
stmt = stmt.where(History.origin_id == filters["origin_id"])
if "thread_id" in filters:
# Include both replies and the parent message (thread_id is the parent uid)
stmt = stmt.where(
or_(
Thread.thread_id == filters["thread_id"],
History.uid == filters["thread_id"],
)
)
if "keyword" in filters:
keyword = filters["keyword"]
# FIXME: "json_each" works with SQLite, but when PostGreSQL support will
# be added, "json_array_elements" may have to be used with it.
subquery = func.json_each(History.extra["keywords"]).table_valued("value")
stmt = stmt.where(exists().where(subquery.c.value == keyword))
if "origin" in filters:
stmt = stmt.where(
History.extra["origin"].as_string() == filters["origin"]
)
if limit is not None:
stmt = stmt.limit(limit)
async with self.session() as session:
result = await session.execute(stmt)
result = result.scalars().unique().all()
if not asc:
result.reverse()
return [h.as_tuple() for h in result]
@aio
async def history_last_messages_metadata_get(
self,
client: SatXMPPEntity,
types: list[str] | None = None,
since: float | None = None,
limit: int | None = None,
) -> dict[str, dict]:
"""Get last message info per conversation peer.
For each peer, returns the timestamp, uid, and message text of the most recent
message.
@param client: Client instance.
@param types: Message types to include.
@param since: Get messages after this timestamp.
@param limit: Max number of peers to return.
@return: Mapping of bare JID strings to {timestamp, uid, message}.
"""
own_userhost = client.jid.userhost()
profile_id = self.profiles[client.profile]
# Filters
where_clauses = [
History.profile_id == profile_id,
or_(History.source == own_userhost, History.dest == own_userhost),
]
if types:
where_clauses.append(History.type.in_(types))
if since is not None:
where_clauses.append(History.timestamp >= since)
# Window function: rank history entries from newest to oldest within each peer
peer_expr = case(
(History.source == own_userhost, History.dest), else_=History.source
)
ranked = (
select(
peer_expr.label("peer"),
History.timestamp.label("last_ts"),
History.uid.label("last_uid"),
over(
func.row_number(),
partition_by=peer_expr,
order_by=text("timestamp DESC, uid DESC"),
).label("row_rank"),
)
.where(*where_clauses)
.subquery()
)
# We only keep results with `row_rank == 1` (i.e. the last message)
stmt = (
select(ranked.c.peer, ranked.c.last_ts, ranked.c.last_uid)
.where(ranked.c.row_rank == 1)
.order_by(desc(ranked.c.last_ts))
)
if limit is not None:
stmt = stmt.limit(limit)
async with self.session() as session:
result = await session.execute(stmt)
rows = {
row[0]: {"timestamp": row[1], "uid": row[2], "message": ""}
for row in result
}
# We fetch message texts separately (only for rows that have uids)
uids = [info["uid"] for info in rows.values() if info["uid"]]
if uids:
msg_stmt = select(Message.history_uid, Message.message).where(
Message.history_uid.in_(uids)
)
msg_result = await session.execute(msg_stmt)
msgs_by_uid = {}
# Pick first non-NULL message per uid
for hist_uid, msg in msg_result:
if hist_uid not in msgs_by_uid and msg:
msgs_by_uid[hist_uid] = msg
# If no non-NULL message has been found, we use an empty string.
for info in rows.values():
info["message"] = msgs_by_uid.get(info["uid"], "")
return rows
@aio
async def unread_counts_get(
self, client: SatXMPPEntity, namespace: str
) -> dict[str, int]:
"""Compute unread message counts per peer.
Joins PrivateInd (displayed state) with History to find the displayed timestamp
per peer, then counts messages after that timestamp.
@param client: Client instance.
@param namespace: PrivateInd namespace for displayed state.
@return: Mapping of peer_jid to unread count.
"""
own_userhost = client.jid.userhost()
profile_id = self.profiles[client.profile]
# Subquery to get timestamp of last seen message per peer JID
last_timestamps_subq = (
select(
PrivateInd.key.label("peer_jid"), History.timestamp.label("msg_timestamp")
)
.join(
History,
and_(PrivateInd.value == History.uid, History.profile_id == profile_id),
)
.where(PrivateInd.namespace == namespace, PrivateInd.profile_id == profile_id)
.subquery()
)
# Be sure to use a peer JID, not our own.
peer_expr = case(
(History.source == own_userhost, History.dest), else_=History.source
)
# Count messages after displayed timestamp per peer
stmt = (
select(peer_expr.label("peer_jid"), func.count().label("cnt"))
.join(last_timestamps_subq, peer_expr == last_timestamps_subq.c.peer_jid)
.where(
History.profile_id == profile_id,
History.timestamp > last_timestamps_subq.c.msg_timestamp,
)
.group_by(peer_expr)
)
async with self.session() as session:
result = await session.execute(stmt)
return {row[0]: row[1] for row in result}
@aio
async def add_to_history(self, data: dict, profile: str) -> None:
"""Store a new message in history
@param data: message data as build by SatMessageProtocol.onMessage
"""
extra = {k: v for k, v in data["extra"].items() if k not in NOT_IN_EXTRA}
messages = [
Message(message=mess, language=lang) for lang, mess in data["message"].items()
]
subjects = [
Subject(subject=mess, language=lang) for lang, mess in data["subject"].items()
]
if "thread" in data["extra"]:
thread = Thread(
thread_id=data["extra"]["thread"],
parent_id=data["extra"].get("thread_parent"),
)
else:
thread = None
try:
async with self.session() as session:
async with session.begin():
session.add(
History(
uid=data["uid"],
origin_id=data["extra"].get("origin_id"),
stanza_id=data["extra"].get("stanza_id"),
update_uid=data["extra"].get("update_uid"),
profile_id=self.profiles[profile],
source_jid=data["from"],
dest_jid=data["to"],
timestamp=data["timestamp"],
received_timestamp=data.get("received_timestamp"),
type=data["type"],
extra=extra,
messages=messages,
subjects=subjects,
thread=thread,
)
)
except IntegrityError as e:
if "unique" in str(e.orig).lower():
log.debug(
f"message {data['uid']!r} is already in history, not storing it again"
)
else:
log.error(f"Can't store message {data['uid']!r} in history: {e}")
except Exception as e:
log.critical(
f"Can't store message, unexpected exception (uid: {data['uid']}): {e}"
)
@aio
async def add_thread(
self,
message_uid: str,
thread_id: str,
parent_id: str | None,
is_retroactive: bool,
) -> None:
"""Add a thread to a message.
@param message_uid: Internal ID of the message (History.uid) to which the thread
is added.
@param thread_id: The thread ID to set.
@param parent_id: The parent message ID in the thread.
@param is_retroactive: Whether the thread ID was added after the message was sent.
Is is used when we want to relate a parent message to a new thread, when the
parent message didn't have a thread ID initially.
"""
async with self.session() as session:
thread = Thread(
thread_id=thread_id,
parent_id=parent_id,
is_retroactive=is_retroactive,
history_uid=message_uid,
)
session.add(thread)
await session.commit()
## Private values
def _get_private_class(self, binary, profile):
"""Get ORM class to use for private values"""
if profile is None:
return PrivateGenBin if binary else PrivateGen
else:
return PrivateIndBin if binary else PrivateInd
@aio
async def get_privates(
self,
namespace: str,
keys: Optional[Iterable[str]] = None,
binary: bool = False,
profile: Optional[str] = None,
) -> Dict[str, Any]:
"""Get private value(s) from databases
@param namespace: namespace of the values
@param keys: keys of the values to get None to get all keys/values
@param binary: True to deserialise binary values
@param profile: profile to use for individual values
None to use general values
@return: gotten keys/values
"""
if keys is not None:
keys = list(keys)
log.debug(
f"getting {'general' if profile is None else 'individual'}"
f"{' binary' if binary else ''} private values from database for namespace "
f"{namespace}{f' with keys {keys!r}' if keys is not None else ''}"
)
cls = self._get_private_class(binary, profile)
stmt = select(cls).filter_by(namespace=namespace)
if keys:
stmt = stmt.where(cls.key.in_(list(keys)))
if profile is not None:
stmt = stmt.filter_by(profile_id=self.profiles[profile])
async with self.session() as session:
result = await session.execute(stmt)
return {p.key: p.value for p in result.scalars()}
@aio
async def set_private_value(
self,
namespace: str,
key: str,
value: Any,
binary: bool = False,
profile: Optional[str] = None,
) -> None:
"""Set a private value in database
@param namespace: namespace of the values
@param key: key of the value to set
@param value: value to set
@param binary: True if it's a binary values
binary values need to be serialised, used for everything but strings
@param profile: profile to use for individual value
if None, it's a general value
"""
cls = self._get_private_class(binary, profile)
values = {"namespace": namespace, "key": key, "value": value}
index_elements = [cls.namespace, cls.key]
if profile is not None:
values["profile_id"] = self.profiles[profile]
index_elements.append(cls.profile_id)
async with self.session() as session:
await session.execute(
insert(cls)
.values(**values)
.on_conflict_do_update(
index_elements=index_elements, set_={cls.value: value}
)
)
await session.commit()
@aio
async def del_private_value(
self,
namespace: str,
key: str,
binary: bool = False,
profile: Optional[str] = None,
) -> None:
"""Delete private value from database
@param category: category of the privateeter
@param key: key of the private value
@param binary: True if it's a binary values
@param profile: profile to use for individual value
if None, it's a general value
"""
cls = self._get_private_class(binary, profile)
stmt = delete(cls).filter_by(namespace=namespace, key=key)
if profile is not None:
stmt = stmt.filter_by(profile_id=self.profiles[profile])
async with self.session() as session:
await session.execute(stmt)
await session.commit()
@aio
async def del_private_namespace(
self, namespace: str, binary: bool = False, profile: Optional[str] = None
) -> None:
"""Delete all data from a private namespace
Be really cautious when you use this method, as all data with given namespace are
removed.
Params are the same as for del_private_value
"""
cls = self._get_private_class(binary, profile)
stmt = delete(cls).filter_by(namespace=namespace)
if profile is not None:
stmt = stmt.filter_by(profile_id=self.profiles[profile])
async with self.session() as session:
await session.execute(stmt)
await session.commit()
## Files
@aio
async def get_files(
self,
client: Optional[SatXMPPEntity],
file_id: Optional[str] = None,
version: Optional[str] = "",
parent: Optional[str] = None,
type_: Optional[str] = None,
file_hash: Optional[str] = None,
hash_algo: Optional[str] = None,
name: Optional[str] = None,
namespace: Optional[str] = None,
mime_type: Optional[str] = None,
public_id: Optional[str] = None,
owner: Optional[jid.JID] = None,
access: Optional[dict] = None,
projection: Optional[List[str]] = None,
unique: bool = False,
) -> List[dict]:
"""Retrieve files with with given filters
@param file_id: id of the file
None to ignore
@param version: version of the file
None to ignore
empty string to look for current version
@param parent: id of the directory containing the files
None to ignore
empty string to look for root files/directories
@param projection: name of columns to retrieve
None to retrieve all
@param unique: if True will remove duplicates
other params are the same as for [set_file]
@return: files corresponding to filters
"""
if projection is None:
projection = [
"id",
"version",
"parent",
"type",
"file_hash",
"hash_algo",
"name",
"size",
"namespace",
"media_type",
"media_subtype",
"public_id",
"created",
"modified",
"owner",
"access",
"extra",
]
stmt = select(*[getattr(File, f) for f in projection])
if unique:
stmt = stmt.distinct()
if client is not None:
stmt = stmt.filter_by(profile_id=self.profiles[client.profile])
else:
if public_id is None:
raise exceptions.InternalError(
"client can only be omitted when public_id is set"
)
if file_id is not None:
stmt = stmt.filter_by(id=file_id)
if version is not None:
stmt = stmt.filter_by(version=version)
if parent is not None:
stmt = stmt.filter_by(parent=parent)
if type_ is not None:
stmt = stmt.filter_by(type=type_)
if file_hash is not None:
stmt = stmt.filter_by(file_hash=file_hash)
if hash_algo is not None:
stmt = stmt.filter_by(hash_algo=hash_algo)
if name is not None:
stmt = stmt.filter_by(name=name)
if namespace is not None:
stmt = stmt.filter_by(namespace=namespace)
if mime_type is not None:
if "/" in mime_type:
media_type, media_subtype = mime_type.split("/", 1)
stmt = stmt.filter_by(media_type=media_type, media_subtype=media_subtype)
else:
stmt = stmt.filter_by(media_type=mime_type)
if public_id is not None:
stmt = stmt.filter_by(public_id=public_id)
if owner is not None:
stmt = stmt.filter_by(owner=owner)
if access is not None:
raise NotImplementedError("Access check is not implemented yet")
# a JSON comparison is needed here
async with self.session() as session:
result = await session.execute(stmt)
return [r._asdict() for r in result]
@aio
async def set_file(
self,
client: SatXMPPEntity,
name: str,
file_id: str,
version: str = "",
parent: str = "",
type_: str = C.FILE_TYPE_FILE,
file_hash: Optional[str] = None,
hash_algo: Optional[str] = None,
size: int = None,
namespace: Optional[str] = None,
mime_type: Optional[str] = None,
public_id: Optional[str] = None,
created: Optional[float] = None,
modified: Optional[float] = None,
owner: Optional[jid.JID] = None,
access: Optional[dict] = None,
extra: Optional[dict] = None,
) -> None:
"""Set a file metadata
@param client: client owning the file
@param name: name of the file (must not contain "/")
@param file_id: unique id of the file
@param version: version of this file
@param parent: id of the directory containing this file
Empty string if it is a root file/directory
@param type_: one of:
- file
- directory
@param file_hash: unique hash of the payload
@param hash_algo: algorithm used for hashing the file (usually sha-256)
@param size: size in bytes
@param namespace: identifier (human readable is better) to group files
for instance, namespace could be used to group files in a specific photo album
@param mime_type: media type of the file, or None if not known/guessed
@param public_id: ID used to server the file publicly via HTTP
@param created: UNIX time of creation
@param modified: UNIX time of last modification, or None to use created date
@param owner: jid of the owner of the file (mainly useful for component)
@param access: serialisable dictionary with access rules. See [memory.memory] for
details
@param extra: serialisable dictionary of any extra data
will be encoded to json in database
"""
if mime_type is None:
media_type = media_subtype = None
elif "/" in mime_type:
media_type, media_subtype = mime_type.split("/", 1)
else:
media_type, media_subtype = mime_type, None
async with self.session() as session:
async with session.begin():
session.add(
File(
id=file_id,
version=version.strip(),
parent=parent,
type=type_,
file_hash=file_hash,
hash_algo=hash_algo,
name=name,
size=size,
namespace=namespace,
media_type=media_type,
media_subtype=media_subtype,
public_id=public_id,
created=time.time() if created is None else created,
modified=modified,
owner=owner,
access=access,
extra=extra,
profile_id=self.profiles[client.profile],
)
)
@aio
async def file_get_used_space(self, client: SatXMPPEntity, owner: jid.JID) -> int:
async with self.session() as session:
result = await session.execute(
select(sum_(File.size)).filter_by(
owner=owner,
type=C.FILE_TYPE_FILE,
profile_id=self.profiles[client.profile],
)
)
return result.scalar_one_or_none() or 0
@aio
async def file_delete(self, file_id: str) -> None:
"""Delete file metadata from the database
@param file_id: id of the file to delete
NOTE: file itself must still be removed, this method only handle metadata in
database
"""
async with self.session() as session:
await session.execute(delete(File).filter_by(id=file_id))
await session.commit()
@aio
async def file_update(
self, file_id: str, column: str, update_cb: Callable[[dict], None]
) -> None:
"""Update a column value using a method to avoid race conditions.
The older value will be retrieved from database, then update_cb will be applied to
update it, and file will be updated checking that older value has not been changed
meanwhile by another user. If it has changed, it tries again a couple of times
before failing.
@param column: column name (only "access" or "extra" are allowed)
@param update_cb: method to update the value of the colum
the method will take older value as argument, and must update it in place
update_cb must not care about serialization,
it get the deserialized data (i.e. a Python object) directly
@raise exceptions.NotFound: there is not file with this id
"""
if column not in ("access", "extra"):
raise exceptions.InternalError("bad column name")
for i in range(5):
async with self.session() as session:
try:
# We use raw SQL to bypass ORM processing and get the literal database
# value, which will be used for comparison on update: if the value has
# been changed between the select and update, the update will fail,
# preventing race condition.
stmt = text(f"SELECT {column} FROM files WHERE id = :file_id")
result = await session.execute(stmt, {"file_id": file_id})
row = result.first()
if not row:
raise NoResultFound()
raw_old_value = row[0]
# We deserialise for the callback.
new_value = {} if raw_old_value is None else json.loads(raw_old_value)
update_cb(new_value)
where_clause = text(f"id = :file_id AND {column} = :raw_old_value")
params = {"file_id": file_id, "raw_old_value": raw_old_value}
update_stmt = text(
f"UPDATE files SET {column} = :new_value WHERE "
+ where_clause.text
)
params["new_value"] = json.dumps(new_value, ensure_ascii=False)
result = await session.execute(update_stmt, params)
await session.commit()
except NoResultFound:
raise exceptions.NotFound
if result.rowcount == 1:
break
log.warning(
"table not updated, probably due to race condition, trying again "
f"({i+1})."
)
else:
raise exceptions.DatabaseError(
f"Can't update file {file_id} due to race condition"
)
@aio
async def get_pubsub_node(
self,
client: SatXMPPEntity,
service: jid.JID,
name: str | None = None,
node_id: int | None = None,
pep: bool | None = None,
with_items: bool = False,
with_subscriptions: bool = False,
with_affiliations: bool = False,
with_parent: bool = False,
with_link: bool = False,
create: bool = False,
create_kwargs: dict | None = None,
) -> PubsubNode | None:
"""Retrieve a PubsubNode from DB
@param service: service hosting the node
@param name: node's name
Can't be used with ``node_id``
@param node_id: node's ID
Can't be used with ``node_id` or ``create``
@param pep: if set, filter on PEP/non-PEP nodes
@param with_items: retrieve items in the same query
@param with_subscriptions: retrieve subscriptions in the same query
@param with_affiliations: retrieve affiliations in the same query
@param with_parent: retrieve parent_node in the same query
@param with_link: retrieve linked_node in the same query
@param create: if the node doesn't exist in DB, create it
@param create_kwargs: keyword arguments to use with ``set_pubsub_node`` if the
node needs to be created.
@raise ValueError: if both name and node_id are provided, or neither.
"""
if node_id is not None:
if name is not None:
raise ValueError('"node_id" and "name" can\'t be used at the same time.')
if create:
raise ValueError("Cannot use create=True with node_id")
async with self.session() as session:
if node_id is not None:
# When querying by ID, we only need it.
stmt = select(PubsubNode).filter_by(id=node_id)
else:
filters = {
"service": service,
"profile_id": self.profiles[client.profile],
"name": name,
}
if pep is not None:
filters["pep"] = pep
stmt = select(PubsubNode).filter_by(**filters)
if with_items:
stmt = stmt.options(joinedload(PubsubNode.items))
if with_subscriptions:
stmt = stmt.options(joinedload(PubsubNode.subscriptions))
if with_affiliations:
stmt = stmt.options(joinedload(PubsubNode.affiliations))
if with_parent:
stmt = stmt.options(joinedload(PubsubNode.parent_node))
if with_link:
stmt = stmt.options(joinedload(PubsubNode.linked_node))
result = await session.execute(stmt)
ret = result.unique().scalar_one_or_none()
if ret is None and create:
assert name is not None
# we auto-create the node
if create_kwargs is None:
create_kwargs = {}
try:
return await self.set_pubsub_node.aio_func(
self, client, service, name, **create_kwargs
)
except IntegrityError as e:
if "unique" in str(e.orig).lower():
# the node may already exist, if it has been created just after
# get_pubsub_node above
log.debug("ignoring UNIQUE constraint error")
return await self.get_pubsub_node.aio_func(
self,
client,
service,
name=name,
with_items=with_items,
with_subscriptions=with_subscriptions,
with_affiliations=with_affiliations,
with_parent=with_parent,
with_link=with_link,
)
else:
raise e
else:
return ret
@aio
async def get_pubsub_nodes(
self,
client: SatXMPPEntity | None,
service: jid.JID | None,
node_name: str | None = None,
linked_node: bool = False,
depth: int = 0,
with_items: bool = False,
with_subscriptions: bool = False,
with_affiliations: bool = False,
with_parent: bool = False,
with_link: bool = False,
) -> list[PubsubNode]:
"""Retrieve pubsub nodes with their hierarchy and linking nodes.
Nodes are returned in a tree, i.e., root nodes are returned, and their children,
linking nodes and items are populated if requested.
@param client: If set, only return nodes of this client profile.
@param service: If set, only return nodes from this service.
@param node_name: If set, start search from this node.
If None, start from all root nodes (nodes without parents).
@param linked_node: If True, populate linking_nodes relationship for each base
node.
@param depth: Maximum depth of child hierarchy to traverse (0 = only starting
nodes).
@param with_items: If True, eagerly load items for each node.
@param with_subscriptions: If True, eagerly load subscriptions for each node.
@param with_affiliations: If True, eagerly load affiliations for each node.
@param with_parent: If True, eagerly load parent_node for each node.
@param with_link: If True, eagerly load linked_node for each node.
@return: List of base nodes matching criteria (root nodes of the request).
"""
# Base criteria that always apply
criteria = []
if client is not None:
criteria.append(PubsubNode.profile_id == self.profiles[client.profile])
if service is not None:
criteria.append(PubsubNode.service == service)
# Add filter for base nodes (not linked to another node)
criteria.append(PubsubNode.linked_node_id.is_(None))
# Prepare query options
# Always include parent_node and linked_node for all nodes when requested
options = []
if depth > 0:
# Load child_nodes when depth > 0 so hierarchy can be traversed
options.append(selectinload(PubsubNode.child_nodes))
if with_parent:
options.append(selectinload(PubsubNode.parent_node))
if with_link:
options.append(joinedload(PubsubNode.linked_node))
if with_items:
options.append(selectinload(PubsubNode.items))
if linked_node:
options.append(selectinload(PubsubNode.linking_nodes))
if with_subscriptions:
options.append(selectinload(PubsubNode.subscriptions))
if with_affiliations:
options.append(selectinload(PubsubNode.affiliations))
async with self.session() as session:
# 1. Determine base query for starting nodes.
if node_name is not None:
# Specific node as starting point (must be a base node).
base_query = select(PubsubNode).filter(
PubsubNode.name == node_name, *criteria
)
else:
# All root base nodes as starting points (nodes without parent that are
# base nodes).
base_query = select(PubsubNode).filter(
PubsubNode.parent_node_id.is_(None), *criteria
)
# Apply eager loading to base query
for option in options:
base_query = base_query.options(option)
# Execute base query first (we need these nodes to build hierarchy).
base_result = await session.execute(base_query)
base_nodes = base_result.unique().scalars().all()
# If no base nodes found, return empty list.
if not base_nodes:
return []
# 2. Build child hierarchy CTE for base nodes (if depth > 0).
if depth > 0:
base_node_ids = [node.id for node in base_nodes]
# Create recursive CTE to find child base nodes.
hierarchy = (
select(PubsubNode.id, literal(0).label("depth"))
.filter(
PubsubNode.id.in_(base_node_ids),
PubsubNode.linked_node_id.is_(None),
)
.cte(name="hierarchy", recursive=True)
)
# Recursive part: get direct children (including nodes with linked_node_id).
children = (
select(PubsubNode.id, hierarchy.c.depth + 1)
.join(hierarchy, PubsubNode.parent_node_id == hierarchy.c.id)
.filter(hierarchy.c.depth < depth)
)
# Combine and create the full hierarchy.
full_hierarchy = hierarchy.union_all(children)
# Final query for all base nodes in the hierarchy.
hierarchy_query = select(PubsubNode).filter(
PubsubNode.id.in_(select(full_hierarchy.c.id))
)
# Apply the same eager loading options.
for option in options:
hierarchy_query = hierarchy_query.options(option)
# Execute hierarchy query.
hierarchy_result = await session.execute(hierarchy_query)
all_base_nodes = hierarchy_result.unique().scalars().all()
# Return only the original base nodes (the root nodes of our query)
# but taken from the fully-loaded hierarchy results.
base_node_ids = {node.id for node in base_nodes}
return [node for node in all_base_nodes if node.id in base_node_ids]
else:
# When depth=0, results are just the base nodes with relationships loaded.
return base_nodes
@aio
async def get_pubsub_node_children(self, node: PubsubNode) -> list[PubsubNode]:
"""Retrieve direct children of a node.
@param node: Parent node.
@return: Child nodes.
"""
async with self.session() as session:
stmt = (
select(PubsubNode)
.where(
PubsubNode.service == node.service,
PubsubNode.parent_node_id == node.id,
)
.order_by(PubsubNode.id)
)
result = await session.execute(stmt)
return result.scalars().all()
@aio
async def get_pep_last_items(
self,
client: SatXMPPEntity,
publishers: list[jid.JID],
node_names: list[str],
access_models: list[AccessModel],
) -> list[tuple[PubsubNode, PubsubItem]]:
"""Get last items for PEP nodes.
This retrieves the last item for each (publisher, node_name) combination in a
single query.
@param client: Client instance for profile filtering.
@param publishers: List of bare JIDs that own PEP nodes.
@param node_names: List of node names to query (e.g., from +notify features).
@param access_models: Allowed node access models.
@return: List of (node, last_item) tuples.
"""
if not publishers or not node_names:
return []
profile_id = self.profiles.get(client.profile)
async with self.session() as session:
# Use window function to rank items by node, ordered by updated DESC, id DESC
# ROW_NUMBER() assigns 1 to the latest item per node
ranked_cte = (
select(
PubsubNode,
PubsubItem,
func.row_number()
.over(
partition_by=PubsubItem.node_id,
order_by=(
desc(PubsubItem.updated),
desc(PubsubItem.id), # Tie-breaker
),
)
.label("row_num"),
)
.join(
PubsubNode,
PubsubItem.node_id == PubsubNode.id,
)
.where(
PubsubNode.profile_id == profile_id,
PubsubNode.service.in_(publishers),
PubsubNode.name.in_(node_names),
PubsubNode.pep == true(),
PubsubNode.access_model.in_(access_models),
)
.cte()
)
# Select only the latest item (row_num = 1) for each node
stmt = select(
ranked_cte.c.PubsubNode,
ranked_cte.c.PubsubItem,
).where(ranked_cte.c.row_num == 1)
result = await session.execute(stmt)
return result.all()
@aio
async def set_pubsub_node(
self,
client: SatXMPPEntity,
service: jid.JID,
name: str,
access_model: AccessModel | None = None,
publish_model: PublishModel | None = None,
analyser: Optional[str] = None,
type_: Optional[str] = None,
subtype: Optional[str] = None,
subscribed: bool = False,
extra: dict | None = None,
items: list[PubsubItem] | None = None,
affiliations: list[PubsubAffiliation] | None = None,
subscriptions: list[PubsubSub] | None = None,
parent_node: PubsubNode | None = None,
linked_node: PubsubNode | None = None,
max_items: int = 0,
pep: bool = False,
) -> PubsubNode:
"""Create a new pubsub node.
@param client: The client creating the node.
@param service: The pubsub service JID.
For PEP nodes (pep=True), this should be the bare JID of the PEP owner.
@param name: The node name.
@param access_model: The access model for the node.
@param publish_model: The publish model for the node.
@param analyser: The analyser for the node.
@param type_: The node type.
@param subtype: The node subtype.
@param subscribed: Whether the creator is subscribed.
@param extra: Extra node metadata.
@param items: Initial items for the node.
@param affiliations: Initial affiliations for the node.
@param subscriptions: Initial subscriptions for the node.
@param parent_node: Parent node (for XEP-0496).
@param linked_node: Linked node (for XEP-0496).
@param max_items: Maximum number of items to persist (0 = unlimited).
@param pep: If True, this is a PEP-scoped node.
For PEP nodes, the service JID should be the bare JID of the PEP owner.
@return: The created node.
@raise ConflictError: If a node with the same name already exists.
"""
node = PubsubNode(
profile_id=self.profiles[client.profile],
service=service,
name=name,
pep=pep,
access_model=access_model,
publish_model=publish_model,
subscribed=subscribed,
analyser=analyser,
type_=type_,
subtype=subtype,
extra=extra,
items=items or [],
affiliations=affiliations or [],
subscriptions=subscriptions or [],
parent_node=parent_node,
linked_node=linked_node,
max_items=max_items,
)
try:
async with self.session() as session:
async with session.begin():
session.add(node)
except IntegrityError as e:
if "unique" in str(e.orig).lower():
raise exceptions.ConflictError(
f"Node {name!r} already exists at {service}"
) from e
raise
return node
@aio
async def update_pubsub_node_sync_state(
self, node: PubsubNode, state: SyncState
) -> None:
async with self.session() as session:
async with session.begin():
await session.execute(
update(PubsubNode)
.filter_by(id=node.id)
.values(
sync_state=state,
sync_state_updated=time.time(),
)
)
@aio
async def update_pubsub_node(self, node: PubsubNode, values: dict) -> None:
async with self.session() as session:
async with session.begin():
await session.execute(
update(PubsubNode).filter_by(id=node.id).values(**values)
)
@aio
async def delete_pubsub_node(
self,
profiles: Optional[List[str]],
services: Optional[List[jid.JID]],
names: Optional[List[str]],
) -> None:
"""Delete items cached for a node
@param profiles: profile names from which nodes must be deleted.
None to remove nodes from ALL profiles
@param services: JIDs of pubsub services from which nodes must be deleted.
None to remove nodes from ALL services
@param names: names of nodes which must be deleted.
None to remove ALL nodes whatever is their names
"""
stmt = delete(PubsubNode)
if profiles is not None:
stmt = stmt.where(
PubsubNode.profile.in_([self.profiles[p] for p in profiles])
)
if services is not None:
stmt = stmt.where(PubsubNode.service.in_(services))
if names is not None:
stmt = stmt.where(PubsubNode.name.in_(names))
async with self.session() as session:
await session.execute(stmt)
await session.commit()
@aio
async def cache_pubsub_items(
self,
client: SatXMPPEntity,
node: PubsubNode,
items: list[domish.Element],
parsed_items: list[dict] | None = None,
) -> None:
"""Add items to database, using an upsert taking care of "updated" field.
Enforces max_items limit by evicting oldest items if necessary.
@param client: the client instance
@param node: the pubsub node
@param items: list of domish.Element items to cache
@param parsed_items: list of parsed item, i.e.; dictionaries with parsed data that
will be stored as JSON. Can be used later to filter requests on JSON values.
"""
if parsed_items is not None and len(items) != len(parsed_items):
raise exceptions.InternalError(
"parsed_items must have the same lenght as items"
)
async with self.session() as session:
async with session.begin():
for idx, item in enumerate(items):
parsed = parsed_items[idx] if parsed_items else None
stmt = (
insert(PubsubItem)
.values(
node_id=node.id,
name=item["id"],
data=item,
parsed=parsed,
)
.on_conflict_do_update(
index_elements=(PubsubItem.node_id, PubsubItem.name),
set_={
PubsubItem.data: item,
PubsubItem.parsed: parsed,
PubsubItem.updated: now(),
},
)
)
await session.execute(stmt)
# Enforce max_items limit if set
if node.max_items and node.max_items > 0:
subquery = (
select(PubsubItem.id)
.filter_by(node_id=node.id)
.order_by(PubsubItem.updated.desc(), PubsubItem.id.desc())
.offset(node.max_items)
).subquery()
delete_stmt = delete(PubsubItem).filter(
PubsubItem.id.in_(select(subquery.c.id))
)
await session.execute(delete_stmt)
await session.commit()
@aio
async def delete_pubsub_items(
self,
node: PubsubNode | None,
items_names: list[str] | None = None,
) -> None:
"""Delete items from a node.
Used for pubsub operations like retract or purge.
@param node: node from which items must be deleted
@param items_names: names of items to delete.
If None, deletes ALL items
"""
stmt = delete(PubsubItem)
if node is not None:
if isinstance(node, list):
stmt = stmt.where(PubsubItem.node_id.in_([n.id for n in node]))
else:
stmt = stmt.filter_by(node_id=node.id)
if items_names is not None:
stmt = stmt.where(PubsubItem.name.in_(items_names))
async with self.session() as session:
await session.execute(stmt)
await session.commit()
@aio
async def get_pubsub_item_count(self, node: PubsubNode) -> int:
"""Count the number of items in a node.
@param node: the pubsub node
@return: the number of items in the node
"""
async with self.session() as session:
async with session.begin():
result = await session.execute(
select(func.count(PubsubItem.id)).filter_by(node_id=node.id)
)
return result.scalar() or 0
@aio
async def purge_pubsub_items(
self,
services: list[jid.JID] | None = None,
names: list[str] | None = None,
types: list[str] | None = None,
subtypes: list[str] | None = None,
profiles: list[str] | None = None,
created_before: datetime | None = None,
updated_before: datetime | None = None,
) -> None:
"""Bulk delete items for maintenance operations.
Deletes items across multiple nodes based on node properties or time criteria.
Used for administrative cleanup tasks.
@param services: filter by pubsub service JIDs
@param names: filter by node names
@param types: filter by node types
@param subtypes: filter by node subtypes
@param profiles: filter by profile names
@param created_before: delete items created before this datetime
@param updated_before: delete items updated before this datetime
"""
stmt = delete(PubsubItem)
node_fields = {
"service": services,
"name": names,
"type_": types,
"subtype": subtypes,
}
if profiles is not None:
node_fields["profile_id"] = [self.profiles[p] for p in profiles]
if any(x is not None for x in node_fields.values()):
sub_q = select(PubsubNode.id)
for col, values in node_fields.items():
if values is None:
continue
sub_q = sub_q.where(getattr(PubsubNode, col).in_(values))
stmt = stmt.where(PubsubItem.node_id.in_(sub_q)).execution_options(
synchronize_session=False
)
if created_before is not None:
stmt = stmt.where(PubsubItem.created < created_before)
if updated_before is not None:
stmt = stmt.where(PubsubItem.updated < updated_before)
async with self.session() as session:
await session.execute(stmt)
await session.commit()
async def _get_items_rsm(
self,
node: PubsubNode,
max_items: int,
before: str | None,
after: str | None,
from_index: int | None,
order: list,
metadata: dict,
) -> tuple[list[PubsubItem], dict]:
"""Get Pubsub Items using RSM
@param node: retrieve items from this node (must be synchronised)
@param max_items: maximum number of items to retrieve
@param before: get items which are before the item with this name
@param after: get items which are after the item with this name
@param from_index: get items starting from this index (None if before/after used)
@param order: SQL order clauses
@param metadata: metadata dict to populate with RSM information
@return: tuple of (items list, metadata dict)
@raise exceptions.NotFound: if before or after references a non-existent item.
"""
# CTE to have result row numbers
row_num_q = select(
PubsubItem.id,
PubsubItem.name,
# row_number starts from 1, but RSM index must start from 0
(func.row_number().over(order_by=order) - 1).label("item_index"),
).filter_by(node_id=node.id)
row_num_cte = row_num_q.cte()
if max_items != 0:
# As we can't simply use PubsubItem.id when we order by modification, we need
# to use row number.
stmt = (
select(row_num_cte.c.item_index, PubsubItem)
.join(row_num_cte, PubsubItem.id == row_num_cte.c.id)
.limit(max_items)
)
if before == "":
# We look for the last page, we need to use descending order for the
# query, and reverse later.
stmt = stmt.order_by(row_num_cte.c.item_index.desc())
elif before or after:
# We use an item name as limit, we find which row number correspond to it.
item_name = before or after
row_num_limit_q = (
select(row_num_cte.c.item_index).where(
row_num_cte.c.name == item_name
)
).scalar_subquery()
# Check if the reference item exists by executing the subquery
async with self.session() as session:
ref_index_result = await session.execute(row_num_limit_q)
ref_index = ref_index_result.scalar_one_or_none()
if ref_index is None:
raise exceptions.NotFound(
f"Item {item_name!r} not found in result set"
)
if before:
# We order by descending for before, we'll reverse after the
# retrieved items.
stmt = stmt.where(row_num_cte.c.item_index < ref_index).order_by(
row_num_cte.c.item_index.desc()
)
elif after:
stmt = stmt.where(row_num_cte.c.item_index > ref_index).order_by(
row_num_cte.c.item_index.asc()
)
else:
# If we reach here, from_index must be set.
assert from_index is not None
stmt = stmt.where(row_num_cte.c.item_index >= from_index).order_by(
row_num_cte.c.item_index.asc()
)
async with self.session() as session:
if max_items == 0:
items = result = []
else:
result = await session.execute(stmt)
result = result.all()
if before is not None:
result.reverse()
items = [row[-1] for row in result]
rows_count = (
await session.execute(row_num_q.with_only_columns(count()))
).scalar_one()
try:
index = result[0][0]
except IndexError:
index = None
try:
first = result[0][1].name
except IndexError:
first = None
last = None
else:
last = result[-1][1].name
metadata["rsm"] = {
k: v
for k, v in {
"index": index,
"count": rows_count,
"first": first,
"last": last,
}.items()
if v is not None
}
metadata["complete"] = (index or 0) + len(result) == rows_count
return items, metadata
@aio
async def get_items(
self,
node: PubsubNode,
max_items: int | None = None,
item_ids: list[str] | None = None,
before: str | None = None,
after: str | None = None,
from_index: int | None = None,
order_by: list[str] | None = None,
desc: bool = True,
force_rsm: bool = False,
) -> tuple[list[PubsubItem], dict]:
"""Get Pubsub Items from cache
@param node: retrieve items from this node (must be synchronised)
@param max_items: maximum number of items to retrieve
@param item_ids: list of specific item IDs to retrieve.
Cannot be used with RSM parameters (before, after, from_index).
@param before: get items which are before the item with this name in given order.
Use empty string (before="") to request the last page.
@param after: get items which are after the item with this name in given order
@param from_index: get items with item index (as defined in RSM spec)
starting from this number
@param order_by: sorting order of items (one of C.ORDER_BY_*)
@param desc: direction of ordering.
If True, ordering is descending.
@param force_rsm: if True, force the use of RSM worklow.
RSM workflow is automatically used if any of before, after or from_index is
used, but if only RSM max_items is used, it won't be used by default. This
parameter let use RSM workflow in this case. Note that in addition to RSM
metadata, the result will not be the same (max_items without RSM will returns
most recent items, i.e. last items in modification order, while max_items with
RSM will return the oldest ones (i.e. first items in modification order).
To be used when max_items is used from RSM
@raise exceptions.BadRequest: Something is wrong with the request.
@raise exceptions.NotFound: if before or after references a non-existent item.
"""
if max_items is not None and max_items < 0:
raise exceptions.BadRequest("max_items must be non-negative")
if from_index is not None and from_index < 0:
raise exceptions.BadRequest("from_index must be non-negative")
metadata = {
"service": node.service,
"node": node.name,
"uri": uri.build_xmpp_uri(
"pubsub",
path=node.service.full(),
node=node.name,
),
}
if max_items is None:
max_items = 20
# ordering
if not order_by:
order_by = [C.ORDER_BY_MODIFICATION]
order = []
for order_type in order_by:
if order_type == C.ORDER_BY_MODIFICATION:
if desc:
order.extend((PubsubItem.updated.desc(), PubsubItem.id.desc()))
else:
order.extend((PubsubItem.updated.asc(), PubsubItem.id.asc()))
elif order_type == C.ORDER_BY_CREATION:
if desc:
order.append(PubsubItem.id.desc())
else:
order.append(PubsubItem.id.asc())
else:
raise exceptions.InternalError(f"Unknown order type {order_type!r}")
use_rsm = any((before is not None, after, from_index is not None))
if force_rsm and not use_rsm:
use_rsm = True
from_index = 0
# item_ids with RSM doesn't make sense
if item_ids is not None and use_rsm:
raise exceptions.BadRequest("item_ids cannot be used with RSM parameters")
if use_rsm:
return await self._get_items_rsm(
node, max_items, before, after, from_index, order, metadata
)
# Query without RSM
stmt = select(PubsubItem).filter_by(node_id=node.id).limit(max_items)
stmt = stmt.order_by(*order)
if item_ids is not None:
stmt = stmt.where(PubsubItem.name.in_(item_ids))
async with self.session() as session:
result = await session.execute(stmt)
result = result.scalars().all()
return result, metadata
@aio
async def get_pubsub_affiliations(
self,
client: SatXMPPEntity,
service: jid.JID | None = None,
entity: jid.JID | None = None,
node_name: str | None = None,
) -> list[tuple[str, jid.JID, Affiliation]]:
"""Retrieve affiliations from the pubsub service.
@param client: Affiliation from this client profile will be retrieved.
@param service: If set, filter to affiliations for this pubsub service.
@param entity: If set, filter to affiliations for this entity.
@param node_name: If set, filter to affiliations for this node.
@return: List of (node_name, entity, affiliation) tuples.
"""
stmt = (
select(
PubsubNode.name, PubsubAffiliation.entity, PubsubAffiliation.affiliation
)
.join(PubsubNode, PubsubAffiliation.node_id == PubsubNode.id)
.where(
PubsubNode.profile_id == self.profiles[client.profile],
)
)
if service is not None:
stmt = stmt.where(PubsubNode.service == service)
if entity is not None:
stmt = stmt.where(PubsubAffiliation.entity == entity)
if node_name is not None:
stmt = stmt.where(PubsubNode.name == node_name)
async with self.session() as session:
result = await session.execute(stmt)
return [(row.name, row.entity, row.affiliation) for row in result.all()]
@aio
async def set_pubsub_affiliations(
self,
client: SatXMPPEntity,
service: jid.JID,
node_name: str,
affiliations: dict[jid.JID, Affiliation | None],
) -> None:
"""Set affiliations for a node.
Uses upsert to insert new affiliations or update existing ones.
Affiliations with value None are removed.
@param client: The profile from this client will be used.
@param service: The pubsub service JID.
@param node_name: The node whose affiliations to set.
@param affiliations: Dict mapping entity JIDs to their new affiliations.
If affiliation is None, the entity's affiliation is removed.
@raise exceptions.NotFound: Node doesn't exist.
"""
node = await self.get_pubsub_node.aio_func(self, client, service, node_name)
if node is None:
raise exceptions.NotFound(f"Node {node_name!r} not found")
async with self.session() as session:
async with session.begin():
to_set = {
entity: aff for entity, aff in affiliations.items() if aff is not None
}
to_delete = [
entity for entity, aff in affiliations.items() if aff is None
]
# Upsert affiliations
if to_set:
for entity, affiliation in to_set.items():
stmt = (
insert(PubsubAffiliation)
.values(
node_id=node.id,
entity=entity,
affiliation=affiliation,
)
.on_conflict_do_update(
index_elements=(
PubsubAffiliation.node_id,
PubsubAffiliation.entity,
),
set_={PubsubAffiliation.affiliation: affiliation},
)
)
await session.execute(stmt)
# Delete affiliations marked as None
if to_delete:
delete_stmt = delete(PubsubAffiliation).where(
PubsubAffiliation.node_id == node.id,
PubsubAffiliation.entity.in_(to_delete),
)
await session.execute(delete_stmt)
await session.commit()
@aio
async def get_pubsub_subscriptions(
self,
client: SatXMPPEntity,
node: PubsubNode | None = None,
service: jid.JID | None = None,
entity: jid.JID | None = None,
node_name: str | None = None,
) -> list[PubsubSub]:
"""Retrieve subscriptions from the pubsub service.
@param client: Subscriptions from this client profile will be retrieved.
@param node: If set, filter to subscriptions for this specific node instance.
Takes precedence over service and node_name. The node will be attached to the
results without additional DB query.
@param service: If set, filter to subscriptions for this pubsub service.
@param entity: If set, filter to subscriptions for this entity (subscriber).
@param node_name: If set, filter to subscriptions for this node.
@return: PubsubSub instances.
@raise exceptions.InternalError: If both node and node_name are provided.
"""
if node is not None and node_name is not None:
raise exceptions.InternalError("Cannot specify both node and node_name")
if node is not None:
stmt = select(PubsubSub).where(PubsubSub.node_id == node.id)
else:
stmt = (
select(PubsubSub)
.join(PubsubNode, PubsubSub.node_id == PubsubNode.id)
.options(joinedload(PubsubSub.node))
.where(
PubsubNode.profile_id == self.profiles[client.profile],
)
)
if service is not None:
stmt = stmt.where(PubsubNode.service == service)
if node_name is not None:
stmt = stmt.where(PubsubNode.name == node_name)
if entity is not None:
stmt = stmt.where(PubsubSub.subscriber == entity)
async with self.session() as session:
result = await session.execute(stmt)
subscriptions = result.scalars().all()
if node is not None:
for sub in subscriptions:
sub.node = node
return subscriptions
@aio
async def set_pubsub_subscriptions(
self,
node: PubsubNode,
subscriptions: dict[jid.JID, SubscriptionState | None],
) -> None:
"""Set subscriptions for a node.
Uses upsert to insert new subscriptions or update existing ones.
Subscriptions with value None are removed.
@param node: The node whose subscriptions to set.
@param subscriptions: Mapping of subscriber JIDs to their new subscription states.
If state is None, the subscription is removed.
"""
async with self.session() as session:
async with session.begin():
to_set = {
subscriber: state
for subscriber, state in subscriptions.items()
if state is not None
}
to_delete = [
subscriber
for subscriber, state in subscriptions.items()
if state is None
]
# Upsert subscriptions
if to_set:
for subscriber, state in to_set.items():
stmt = (
insert(PubsubSub)
.values(
node_id=node.id,
subscriber=subscriber,
state=state,
)
.on_conflict_do_update(
index_elements=(
PubsubSub.node_id,
PubsubSub.subscriber,
),
set_={PubsubSub.state: state},
)
)
await session.execute(stmt)
# Delete subscriptions marked as None
if to_delete:
delete_stmt = delete(PubsubSub).where(
PubsubSub.node_id == node.id,
PubsubSub.subscriber.in_(to_delete),
)
await session.execute(delete_stmt)
await session.commit()
def _get_sqlite_path(self, path: List[Union[str, int]]) -> str:
"""generate path suitable to query JSON element with SQLite"""
return f"${''.join(f'[{p}]' if isinstance(p, int) else f'.{p}' for p in path)}"
@aio
async def search_pubsub_items(
self,
query: dict,
) -> Tuple[List[PubsubItem]]:
"""Search for pubsub items in cache
@param query: search terms. Keys can be:
:fts (str):
Full-Text Search query. Currently SQLite FT5 engine is used, its query
syntax can be used, see `FTS5 Query documentation
<https://sqlite.org/fts5.html#full_text_query_syntax>`_
:profiles (list[str]):
filter on nodes linked to those profiles
:nodes (list[str]):
filter on nodes with those names
:services (list[jid.JID]):
filter on nodes from those services
:types (list[str|None]):
filter on nodes with those types. None can be used to filter on nodes with
no type set
:subtypes (list[str|None]):
filter on nodes with those subtypes. None can be used to filter on nodes with
no subtype set
:names (list[str]):
filter on items with those names
:parsed (list[dict]):
Filter on a parsed data field. The dict must contain 3 keys: ``path``
which is a list of str or int giving the path to the field of interest
(str for a dict key, int for a list index), ``operator`` with indicate the
operator to use to check the condition, and ``value`` which depends of
field type and operator.
See documentation for details on operators (it's currently explained at
``doc/libervia-cli/pubsub_cache.rst`` in ``search`` command
documentation).
:order-by (list[dict]):
Indicates how to order results. The dict can contain either a ``order``
for a well-know order or a ``path`` for a parsed data field path
(``order`` and ``path`` can't be used at the same time), an an optional
``direction`` which can be ``asc`` or ``desc``. See documentation for
details on well-known orders (it's currently explained at
``doc/libervia-cli/pubsub_cache.rst`` in ``search`` command
documentation).
:index (int):
starting index of items to return from the query result. It's translated
to SQL's OFFSET
:limit (int):
maximum number of items to return. It's translated to SQL's LIMIT.
@result: found items (the ``node`` attribute will be filled with suitable
PubsubNode)
"""
# TODO: FTS and parsed data filters use SQLite specific syntax
# when other DB engines will be used, this will have to be adapted
stmt = select(PubsubItem)
# Full-Text Search
fts = query.get("fts")
if fts:
fts_select = (
text("SELECT rowid, rank FROM pubsub_items_fts(:fts_query)")
.bindparams(fts_query=fts)
.columns(rowid=Integer)
.subquery()
)
stmt = stmt.select_from(fts_select).outerjoin(
PubsubItem, fts_select.c.rowid == PubsubItem.id
)
# node related filters
profiles = query.get("profiles")
if profiles or any(
query.get(k) for k in ("nodes", "services", "types", "subtypes")
):
stmt = stmt.join(PubsubNode).options(contains_eager(PubsubItem.node))
if profiles:
try:
stmt = stmt.where(
PubsubNode.profile_id.in_(self.profiles[p] for p in profiles)
)
except KeyError as e:
raise exceptions.ProfileUnknownError(
f"This profile doesn't exist: {e.args[0]!r}"
)
for key, attr in (
("nodes", "name"),
("services", "service"),
("types", "type_"),
("subtypes", "subtype"),
):
value = query.get(key)
if not value:
continue
if key in ("types", "subtypes") and None in value:
# NULL can't be used with SQL's IN, so we have to add a condition with
# IS NULL, and use a OR if there are other values to check
value.remove(None)
condition = getattr(PubsubNode, attr).is_(None)
if value:
condition = or_(getattr(PubsubNode, attr).in_(value), condition)
else:
condition = getattr(PubsubNode, attr).in_(value)
stmt = stmt.where(condition)
else:
stmt = stmt.options(selectinload(PubsubItem.node))
# names
names = query.get("names")
if names:
stmt = stmt.where(PubsubItem.name.in_(names))
# parsed data filters
parsed = query.get("parsed", [])
for filter_ in parsed:
try:
path = filter_["path"]
operator = filter_["op"]
value = filter_["value"]
except KeyError as e:
raise ValueError(
f'missing mandatory key {e.args[0]!r} in "parsed" filter'
)
try:
op_attr = OP_MAP[operator]
except KeyError:
raise ValueError(f"invalid operator: {operator!r}")
sqlite_path = self._get_sqlite_path(path)
if operator in ("overlap", "ioverlap", "disjoint", "idisjoint"):
col = literal_column("json_each.value")
if operator[0] == "i":
col = func.lower(col)
value = [str(v).lower() for v in value]
condition = (
select(1)
.select_from(func.json_each(PubsubItem.parsed, sqlite_path))
.where(col.in_(value))
).scalar_subquery()
if operator in ("disjoint", "idisjoint"):
condition = condition.is_(None)
stmt = stmt.where(condition)
elif operator == "between":
try:
left, right = value
except (ValueError, TypeError):
raise ValueError(
_(
'invalid value for "between" filter, you must use a 2 items '
"array: {value!r}"
).format(value=value)
)
col = func.json_extract(PubsubItem.parsed, sqlite_path)
stmt = stmt.where(col.between(left, right))
else:
# we use func.json_extract instead of generic JSON way because SQLAlchemy
# add a JSON_QUOTE to the value, and we want SQL value
col = func.json_extract(PubsubItem.parsed, sqlite_path)
stmt = stmt.where(getattr(col, op_attr)(value))
# order
order_by = query.get("order-by") or [{"order": "creation"}]
for order_data in order_by:
order, path = order_data.get("order"), order_data.get("path")
if order and path:
raise ValueError(
_(
'"order" and "path" can\'t be used at the same time in '
'"order-by" data'
)
)
if order:
if order == "creation":
col = PubsubItem.id
elif order == "modification":
col = PubsubItem.updated
elif order == "item_id":
col = PubsubItem.name
elif order == "rank":
if not fts:
raise ValueError(
"'rank' order can only be used with Full-Text Search (fts)"
)
col = literal_column("rank")
else:
raise NotImplementedError(f"Unknown {order!r} order")
else:
# we have a JSON path
# sqlite_path = self._get_sqlite_path(path)
col = PubsubItem.parsed[path]
direction = order_data.get("direction", "ASC").lower()
if not direction in ("asc", "desc"):
raise ValueError(f"Invalid order-by direction: {direction!r}")
stmt = stmt.order_by(getattr(col, direction)())
# offset, limit
index = query.get("index")
if index:
stmt = stmt.offset(index)
limit = query.get("limit")
if limit:
stmt = stmt.limit(limit)
async with self.session() as session:
result = await session.execute(stmt)
return result.scalars().all()
# Notifications
@aio
async def add_notification(
self,
client: Optional[SatXMPPEntity],
type_: NotificationType,
body_plain: str,
body_rich: Optional[str] = None,
title: Optional[str] = None,
requires_action: bool = False,
priority: NotificationPriority = NotificationPriority.MEDIUM,
expire_at: Optional[float] = None,
extra: Optional[dict] = None,
) -> Notification:
"""Add a new notification to the DB.
@param client: client associated with the notification. If None, the notification
will be global.
@param type_: type of the notification.
@param body_plain: plain text body.
@param body_rich: rich text (XHTML) body.
@param title: optional title.
@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.
@return: created Notification
"""
profile_id = self.profiles[client.profile] if client else None
notification = Notification(
profile_id=profile_id,
type=type_,
body_plain=body_plain,
body_rich=body_rich,
requires_action=requires_action,
priority=priority,
expire_at=expire_at,
title=title,
extra_data=extra,
status=NotificationStatus.new,
)
async with self.session() as session:
async with session.begin():
session.add(notification)
return notification
@aio
async def update_notification(
self, client: SatXMPPEntity, notification_id: int, **kwargs
) -> None:
"""Update an existing notification.
@param client: client associated with the notification.
@param notification_id: ID of the notification to update.
"""
profile_id = self.profiles[client.profile]
async with self.session() as session:
await session.execute(
update(Notification)
.where(
and_(
Notification.profile_id == profile_id,
Notification.id == notification_id,
)
)
.values(**kwargs)
)
await session.commit()
@aio
async def get_notifications(
self,
client: SatXMPPEntity,
type_: Optional[NotificationType] = None,
status: Optional[NotificationStatus] = None,
requires_action: Optional[bool] = None,
min_priority: Optional[int] = None,
) -> List[Notification]:
"""Retrieve all notifications for a given profile with optional filters.
@param client: client associated with the notifications.
@param type_: filter by type of the notification.
@param status: filter by status of the notification.
@param requires_action: filter by notifications that require user action.
@param min_priority: filter by minimum priority value.
@return: list of matching Notification instances.
"""
profile_id = self.profiles[client.profile]
filters = [
or_(Notification.profile_id == profile_id, Notification.profile_id.is_(None))
]
if type_:
filters.append(Notification.type == type_)
if status:
filters.append(Notification.status == status)
if requires_action is not None:
filters.append(Notification.requires_action == requires_action)
if min_priority:
filters.append(Notification.priority >= min_priority)
async with self.session() as session:
result = await session.execute(
select(Notification).where(and_(*filters)).order_by(Notification.id)
)
return result.scalars().all()
@aio
async def delete_notification(
self, client: Optional[SatXMPPEntity], notification_id: str
) -> None:
"""Delete a notification by its profile and id.
@param client: client associated with the notification. If None, profile_id will be NULL.
@param notification_id: ID of the notification to delete.
"""
profile_id = self.profiles[client.profile] if client else None
async with self.session() as session:
await session.execute(
delete(Notification).where(
and_(
Notification.profile_id == profile_id,
Notification.id == int(notification_id),
)
)
)
await session.commit()
@aio
async def clean_expired_notifications(
self, client: Optional[SatXMPPEntity], limit_timestamp: Optional[float] = None
) -> None:
"""Cleans expired notifications and older profile-specific notifications.
- Removes all notifications where the expiration timestamp has passed,
irrespective of their profile.
- If a limit_timestamp is provided, removes older notifications with a profile set
(i.e., not global notifications) that do not require user action. If client is
provided, only remove notification for this profile.
@param client: if provided, only expire notification for this client (in addition
to truly expired notifications for everybody).
@param limit_timestamp: Timestamp limit for older notifications. If None, only
truly expired notifications are removed.
"""
# Delete truly expired notifications
expired_condition = Notification.expire_at < time.time()
# Delete older profile-specific notifications (created before the limit_timestamp)
if client is None:
profile_condition = Notification.profile_id.isnot(None)
else:
profile_condition = Notification.profile_id == self.profiles[client.profile]
older_condition = and_(
profile_condition,
Notification.timestamp < limit_timestamp if limit_timestamp else False,
Notification.requires_action == False,
)
# Combine the conditions
conditions = or_(expired_condition, older_condition)
async with self.session() as session:
await session.execute(delete(Notification).where(conditions))
await session.commit()
Reference in a new issue View git blame Copy permalink