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

component Pubsub: initial implementation:

Browse source 
Base component implementation, handle node creation, listing, configuration and metadata.

rel 476
This commit is contained in:
Goffi 2026-04-04 00:49:04 +02:00
commit 9cf71f2119
2 changed files with 596 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1408,9 +1408,16 @@ class Storage:
parent_node=parent_node,
linked_node=linked_node,
)
async with self.session() as session:
async with session.begin():
session.add(node)
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
@ -1428,6 +1435,18 @@ class Storage:
)
)
@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,
@ -1462,10 +1481,17 @@ class Storage:
self,
client: SatXMPPEntity,
node: PubsubNode,
items: List[domish.Element],
parsed_items: Optional[List[dict]] = None,
items: list[domish.Element],
parsed_items: list[dict] | None = None,
) -> None:
"""Add items to database, using an upsert taking care of "updated" field"""
"""Add items to database, using an upsert taking care of "updated" field
@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"

564
libervia/backend/plugins/plugin_comp_pubsub/__init__.py Normal file
View file

@ -0,0 +1,564 @@
#!/usr/bin/env python3
# Libervia Pubsub Component
# 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/>.
from dataclasses import dataclass, field, fields
from enum import StrEnum, auto
from typing import Literal, cast
import shortuuid
from twisted.internet import defer
from twisted.words.protocols.jabber import error, jid
from twisted.words.protocols.jabber.xmlstream import XMPPHandler
from twisted.words.xish import domish
from tx_xmpp import data_form, disco, itx_xmpp
from tx_xmpp import pubsub, rsm
from zope.interface import implementer
from libervia.backend import G
from libervia.backend.core import exceptions
from libervia.backend.core.constants import Const as C
from libervia.backend.core.core_types import SatXMPPComponent, SatXMPPEntity
from libervia.backend.core.i18n import _
from libervia.backend.core.log import getLogger
from libervia.backend.memory.sqla_mapping import (
AccessModel,
Affiliation,
PublishModel,
PubsubAffiliation,
)
from libervia.backend.plugins.plugin_xep_0167 import NS_AV_CONFERENCES
from libervia.backend.plugins.plugin_xep_0060 import NodeMetadata, XEP_0060
from libervia.backend.tools.config import ComponentSettings
from libervia.backend.tools.utils import ensure_deferred
log = getLogger(__name__)
IMPORT_NAME = "pubsub"
NAME = "Libervia Pubsub"
PLUGIN_INFO = {
C.PI_NAME: NAME,
C.PI_IMPORT_NAME: IMPORT_NAME,
C.PI_MODES: [C.PLUG_MODE_COMPONENT],
C.PI_TYPE: C.PLUG_TYPE_ENTRY_POINT,
C.PI_PROTOCOLS: [],
C.PI_DEPENDENCIES: ["XEP-0060"],
C.PI_RECOMMENDATIONS: [],
C.PI_MAIN: "PubsubComponent",
C.PI_HANDLER: C.BOOL_TRUE,
C.PI_DESCRIPTION: _("Libervia advanced Pubsub service."),
}
@dataclass
class PubsubNodeConfig:
max_items: int | Literal["max"] = field(
default="max", metadata={"label": "Maximum number of items to persist"}
)
access_model: AccessModel = field(
default=AccessModel.open,
metadata={"label": "Who may subscribe and retrieve items"},
)
publish_model: PublishModel = field(
default=PublishModel.publishers, metadata={"label": "The publisher model"}
)
title: str | None = field(
default=None, metadata={"label": "A short name for the node"}
)
description: str | None = field(
default=None, metadata={"label": "A description of the node"}
)
_field_var_mapping = {
"access_model": "pubsub#access_model",
"publish_model": "pubsub#publish_model",
"max_items": "pubsub#max_items",
"title": "pubsub#title",
"description": "pubsub#description",
}
@classmethod
def get_field_definitions(cls) -> dict[str, dict]:
"""Return field definitions for Tx-XMPP configuration forms.
This method dynamically builds the schema by introspecting dataclass fields
and their metadata.
@return: A dict mapping pubsub field vars to their definitions.
"""
definitions: dict = {}
for field in fields(cls):
field_name = field.name
try:
pubsub_var = cls._field_var_mapping[field_name]
except KeyError:
log.error("Missing field mapping for {field_name!r}, ignoring.")
continue
label = field.metadata.get("label", pubsub_var)
match field_name:
case "access_model" | "publish_model":
enum_type = (
AccessModel if field_name == "access_model" else PublishModel
)
definitions[pubsub_var] = {
"type": "list-single",
"label": label,
"options": {value.value: None for value in enum_type},
}
case _:
definitions[pubsub_var] = {
"type": "text-single",
"label": label,
}
return definitions
class PubsubSettings(ComponentSettings, plugin_info=PLUGIN_INFO):
pass
class PubsubResource(pubsub.PubSubResource):
def __init__(self, service: "PubsubService") -> None:
super().__init__()
self.service = service
@property
def client(self) -> SatXMPPComponent:
return self.service.client
def getInfo(
self,
requestor: jid.JID,
service: jid.JID,
nodeIdentifier: str,
) -> defer.Deferred[dict | None]:
return defer.ensureDeferred(self.get_info(requestor, service, nodeIdentifier))
async def get_info(
self,
requestor: jid.JID,
service: jid.JID,
nodeIdentifier: str,
) -> dict | None:
"""Get node information for disco#info.
@param requestor: The entity the request originated from.
@param service: The publish-subscribe service entity.
@param nodeIdentifier: Identifier of the node to get info for.
@return: Node type and metadata, as expected by ``Tx-XMPP``.
"""
node = await G.storage.get_pubsub_node(self.client, service, nodeIdentifier)
if node is None:
return None
metadata = NodeMetadata(
access_model=str(node.access_model) if node.access_model else None,
publish_model=str(node.publish_model) if node.publish_model else None,
)
# We convert to the format expected by Tx-XMPP.
form = metadata.to_data_form()
metadata_list = [
{
"var": field.var,
"type": field.fieldType,
"value": field.value,
"label": field.label,
}
for field in form.fieldList
]
return {
"type": "leaf",
"meta-data": metadata_list if metadata_list else None,
}
def getNodes(
self, requestor: jid.JID, service: jid.JID, nodeIdentifier: str
) -> defer.Deferred[list[str]]:
return defer.ensureDeferred(self._get_nodes(requestor, service, nodeIdentifier))
async def _get_nodes(
self, requestor: jid.JID, service: jid.JID, node_id: str
) -> list[str]:
"""Get list of child nodes for a given node (async implementation).
@param requestor: The entity the request originated from.
@param service: The publish-subscribe service entity.
@param node_id: Identifier of the node to request the children for.
@return: A list of node identifiers.
"""
if node_id:
# We only return results for requests without a specific node
return []
nodes = await G.storage.get_pubsub_nodes(self.client, self.client.jid)
return [cast(str, node.name) for node in nodes]
def getConfigurationOptions(self) -> dict[str, dict]:
"""Return the available configuration options for pubsub nodes.
@return: A map field vars to their definition in the format expected by Tx-XMPP.
"""
return PubsubNodeConfig.get_field_definitions()
def publish(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._publish(request))
async def _publish(self, request: rsm.PubSubRequest) -> None:
return
def subscribe(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._subscribe(request))
async def _subscribe(self, request: rsm.PubSubRequest) -> None:
return
def unsubscribe(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._unsubscribe(request))
async def _unsubscribe(self, request: rsm.PubSubRequest) -> None:
return
def subscriptions(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._subscriptions(request))
async def _subscriptions(self, request: rsm.PubSubRequest) -> None:
return
def affiliations(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._affiliations(request))
async def _affiliations(self, request: rsm.PubSubRequest) -> None:
return
def create(self, request: rsm.PubSubRequest) -> defer.Deferred[str]:
return defer.ensureDeferred(self._create(request))
async def _create(self, request: rsm.PubSubRequest) -> str:
"""Create a new pubsub node.
@param request: Node creation request.
@return: The identifier of the created node.
"""
client = self.client
assert request.sender is not None
requestor_jid = request.sender.userhostJID()
if not client.is_local(requestor_jid):
raise error.StanzaError("forbidden")
node_id = request.nodeIdentifier or f"generic/{shortuuid.uuid()}"
config = PubsubNodeConfig()
if request.options is not None:
access_model_value = request.options.get(XEP_0060.OPT_ACCESS_MODEL)
if access_model_value:
try:
config.access_model = AccessModel(access_model_value)
except ValueError:
pass
publish_model_value = request.options.get(XEP_0060.OPT_PUBLISH_MODEL)
if publish_model_value:
try:
config.publish_model = PublishModel(publish_model_value)
except ValueError:
pass
owner_affiliation = PubsubAffiliation(
entity=requestor_jid,
affiliation=Affiliation.owner,
)
try:
await G.storage.set_pubsub_node(
client=client,
service=client.jid,
name=node_id,
access_model=config.access_model,
publish_model=config.publish_model,
affiliations=[owner_affiliation],
)
except exceptions.ConflictError:
raise error.StanzaError("conflict", text=f"node {node_id!r} already exists.")
return node_id
def default(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._default(request))
async def _default(self, request: rsm.PubSubRequest) -> None:
return
def configureGet(self, request: rsm.PubSubRequest) -> defer.Deferred[dict]:
return defer.ensureDeferred(self._configureGet(request))
async def _configureGet(self, request: rsm.PubSubRequest) -> dict:
"""Retrieve the configuration for a specific node.
@param request: The configuration request.
@return: Current configuration values for the node.
@raise StanzaError: with "item_not_found" if the node does not exist
@raise StanzaError: with "forbidden" if the requestor is not an owner or admin
"""
client = self.client
assert request.sender is not None
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
node = await G.storage.get_pubsub_node(
client, client.jid, request.nodeIdentifier, with_affiliations=True
)
if node is None:
raise error.StanzaError("item_not_found", text="Node does not exist")
# Only admins and owner can get configuration.
if not G.host.memory.is_admin_jid(requestor_jid):
affiliation = next(
(
aff_data.affiliation
for aff_data in node.affiliations
if aff_data.entity == requestor_jid
),
None,
)
# Only owners can configure nodes
if affiliation != Affiliation.owner:
raise error.StanzaError("forbidden", text="Insufficient privileges")
config_values: dict[str, str | bool | list[str]] = {}
if node.extra:
if "title" in node.extra:
config_values["pubsub#title"] = node.extra["title"]
if "description" in node.extra:
config_values["pubsub#description"] = node.extra["description"]
if node.access_model:
config_values["pubsub#access_model"] = str(node.access_model)
if node.publish_model:
config_values["pubsub#publish_model"] = str(node.publish_model)
# TODO: Add max_items when it's stored in the database
# if node.max_items is not None:
# config_values["pubsub#max_items"] = (
# "max" if node.max_items == -1 else str(node.max_items)
# )
return config_values
def configureSet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._configureSet(request))
async def _configureSet(self, request: rsm.PubSubRequest) -> None:
"""Set the configuration of a node.
@raise StanzaError: with "item_not_found" if the node does not exist
@raise StanzaError: with "forbidden" if the requestor is not an owner or admin
@raise StanzaError: with "not_acceptable" if configuration values are invalid
"""
client = self.client
assert request.sender is not None
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
node = await G.storage.get_pubsub_node(
client, client.jid, request.nodeIdentifier, with_affiliations=True
)
if node is None:
raise error.StanzaError("item_not_found", text="Node does not exist")
# Only admins and owner can modifify configuration.
if not G.host.memory.is_admin_jid(requestor_jid):
affiliation = next(
(
aff_data.affiliation
for aff_data in node.affiliations
if aff_data.entity == requestor_jid
),
None,
)
if affiliation != Affiliation.owner:
raise error.StanzaError("forbidden", text="Insufficient privileges")
if request.options is None:
return
config = PubsubNodeConfig()
access_model_value = request.options.get(XEP_0060.OPT_ACCESS_MODEL)
if access_model_value:
try:
config.access_model = AccessModel(access_model_value)
except ValueError:
raise error.StanzaError(
"not-acceptable",
text=f"Invalid access_model value: {access_model_value!r}",
)
publish_model_value = request.options.get(XEP_0060.OPT_PUBLISH_MODEL)
if publish_model_value:
try:
config.publish_model = PublishModel(publish_model_value)
except ValueError:
raise error.StanzaError(
"not-acceptable",
text=f"Invalid publish_model value: {publish_model_value!r}",
)
title_value = request.options.get("pubsub#title")
description_value = request.options.get("pubsub#description")
extra = node.extra.copy() if node.extra else {}
update_extra = False
if title_value is not None:
extra["title"] = title_value
update_extra = True
if description_value is not None:
extra["description"] = description_value
update_extra = True
update_values = {}
if config.access_model != node.access_model:
update_values["access_model"] = config.access_model
if config.publish_model != node.publish_model:
update_values["publish_model"] = config.publish_model
if update_extra:
update_values["extra"] = extra
if update_values:
await G.storage.update_pubsub_node(node, update_values)
def items(
self,
request: rsm.PubSubRequest,
) -> defer.Deferred[tuple[list[domish.Element], rsm.RSMResponse | None]]:
return defer.ensureDeferred(self._items(request))
async def _items(
self, request: rsm.PubSubRequest
) -> tuple[list[domish.Element], rsm.RSMResponse | None]:
client = self.client
assert request.sender is not None
requestor_jid = request.sender.userhostJID()
if not client.is_local(requestor_jid):
raise error.StanzaError("forbidden")
return [], None
def retract(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._retract(request))
async def _retract(self, request: rsm.PubSubRequest) -> None:
return
def purge(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._purge(request))
async def _purge(self, request: rsm.PubSubRequest) -> None:
return
def delete(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._delete(request))
async def _delete(self, request: rsm.PubSubRequest) -> None:
return
def affiliationsGet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._affiliationsGet(request))
async def _affiliationsGet(self, request: rsm.PubSubRequest) -> None:
return
def affiliationsSet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._affiliationsSet(request))
async def _affiliationsSet(self, request: rsm.PubSubRequest) -> None:
return
def subscriptionsGet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._subscriptionsGet(request))
async def _subscriptionsGet(self, request: rsm.PubSubRequest) -> None:
return
def subscriptionsSet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
return defer.ensureDeferred(self._subscriptionsSet(request))
async def _subscriptionsSet(self, request: rsm.PubSubRequest) -> None:
return
class PubsubService(rsm.PubSubService):
def __init__(self, component: "PubsubComponent"):
super().__init__(resource=PubsubResource(self))
self.component = component
self.discoidentity = {
"category": "pubsub",
"type": "service",
"name": "Libervia Pubsub",
}
@property
def client(self) -> SatXMPPComponent:
return self.component._client
class PubsubComponent:
IMPORT_NAME = IMPORT_NAME
verbose = 0
def __init__(self, host):
self._clientt: SatXMPPComponent | None = None
self.pubsub_service = PubsubService(self)
def client(self) -> SatXMPPComponent:
assert self._client is not None
return self._client
def get_handler(self, __) -> PubsubService:
return self.pubsub_service
def profile_connecting(self, client: SatXMPPEntity) -> None:
self._client = cast(SatXMPPComponent, client)
@implementer(itx_xmpp.IDisco)
class PubsubHandler(XMPPHandler):
def getDiscoInfo(
self, requestor, target, nodeIdentifier=""
) -> list[disco.DiscoFeature]:
return [disco.DiscoFeature(NS_AV_CONFERENCES)]