component pubsub: Implement remaining methods:
- `retract`, `purge` and `delete` are now implemented - `features` list has been added - `publish` now returns the Item ID when it is generated - `subscribers` publish model is handled. rel 476
This commit is contained in:
Goffi
parent
572662653f
commit
e7917b1b63
1 changed files with 226 additions and 43 deletions
|
|
@ -144,6 +144,27 @@ class PubsubSettings(ComponentSettings, plugin_info=PLUGIN_INFO):
|
|||
|
||||
|
||||
class PubsubResource(pubsub.PubSubResource):
|
||||
"""PubSub resource implementation for Libervia Backend.
|
||||
|
||||
This resource handles all pubsub operations including node creation,
|
||||
item publishing, subscriptions, affiliations, and node management.
|
||||
"""
|
||||
|
||||
features: list[str] = [
|
||||
"config-node",
|
||||
"create-nodes",
|
||||
"delete-nodes",
|
||||
"item-ids",
|
||||
"meta-data",
|
||||
"publish",
|
||||
"purge-nodes",
|
||||
"retract-items",
|
||||
"retrieve-affiliations",
|
||||
"retrieve-default",
|
||||
"retrieve-items",
|
||||
"retrieve-subscriptions",
|
||||
"subscribe",
|
||||
]
|
||||
|
||||
def __init__(self, service: "PubsubService") -> None:
|
||||
super().__init__()
|
||||
|
|
@ -230,16 +251,20 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
"""
|
||||
return PubsubNodeConfig.get_field_definitions()
|
||||
|
||||
def publish(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]:
|
||||
def publish(
|
||||
self, request: pubsub.PubSubRequest
|
||||
) -> defer.Deferred[None | domish.Element]:
|
||||
return defer.ensureDeferred(self._publish(request))
|
||||
|
||||
async def _publish(self, request: pubsub.PubSubRequest) -> None:
|
||||
async def _publish(self, request: pubsub.PubSubRequest) -> None | domish.Element:
|
||||
"""Publish items to a node.
|
||||
|
||||
@raise StanzaError: with "bad_request" if no items are provided
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@param request: The publish request.
|
||||
@return: A pubsub element containing the published items.
|
||||
@raise StanzaError: with "bad-request" if no items are provided
|
||||
@raise StanzaError: with "item-not-found" if the node does not exist
|
||||
@raise StanzaError: with "forbidden" if the requestor lacks publish privileges
|
||||
@raise StanzaError: with "not_allowed" if item format is invalid
|
||||
@raise StanzaError: with "not-allowed" if item format is invalid
|
||||
"""
|
||||
client = self.client
|
||||
assert request.sender is not None
|
||||
|
|
@ -248,7 +273,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
assert request.nodeIdentifier is not None
|
||||
|
||||
if request.items is None or len(request.items) == 0:
|
||||
raise error.StanzaError("bad_request", text="no item to publish")
|
||||
raise error.StanzaError("bad-request", text="no item to publish")
|
||||
|
||||
node = await G.storage.get_pubsub_node(
|
||||
client,
|
||||
|
|
@ -259,7 +284,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
raise error.StanzaError("item-not-found", text="Node does not exist")
|
||||
|
||||
affiliation = next(
|
||||
(
|
||||
|
|
@ -277,20 +302,23 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
raise error.StanzaError("forbidden", text="Insufficient privileges")
|
||||
elif publish_model == PublishModel.subscribers:
|
||||
if affiliation not in (Affiliation.owner, Affiliation.publisher):
|
||||
# FIXME
|
||||
raise NotImplementedError("FIXME: need to handle subscribers")
|
||||
# is_subscribed = any(
|
||||
# sub.subscriber == requestor_jid and sub.state is not None
|
||||
# for sub in node.subscriptions
|
||||
# )
|
||||
# if not is_subscribed:
|
||||
# raise error.StanzaError("forbidden", text="Insufficient privileges")
|
||||
is_subscribed = any(
|
||||
sub.subscriber == requestor_jid
|
||||
and sub.state == SubscriptionState.SUBSCRIBED
|
||||
for sub in node.subscriptions
|
||||
)
|
||||
if not is_subscribed:
|
||||
raise error.StanzaError("forbidden", text="Insufficient privileges")
|
||||
|
||||
ret_pubsub_elt = domish.Element((NS_PUBSUB_EVENT, "pubsub"))
|
||||
ret_publish_elt = ret_pubsub_elt.addElement("publish")
|
||||
ret_publish_elt["node"] = request.nodeIdentifier
|
||||
|
||||
items_data = []
|
||||
for item in request.items:
|
||||
if item.name != "item" or item.uri != pubsub.NS_PUBSUB:
|
||||
raise error.StanzaError(
|
||||
"not_allowed", text=f"Invalid item: {item.toXml()}"
|
||||
"not-allowed", text=f"Invalid item: {item.toXml()}"
|
||||
)
|
||||
|
||||
item["publisher"] = request.sender.full()
|
||||
|
|
@ -298,12 +326,17 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
if not item.getAttribute("id"):
|
||||
item["id"] = f"{shortuuid.uuid()}"
|
||||
|
||||
ret_item_elt = ret_publish_elt.addElement("item")
|
||||
ret_item_elt["id"] = item["id"]
|
||||
|
||||
items_data.append(item)
|
||||
|
||||
await G.storage.cache_pubsub_items(client, node, items_data)
|
||||
|
||||
await self._send_notifications(node, "items", items_data)
|
||||
|
||||
return ret_pubsub_elt
|
||||
|
||||
def subscribe(
|
||||
self, request: pubsub.PubSubRequest
|
||||
) -> defer.Deferred[pubsub.Subscription]:
|
||||
|
|
@ -315,7 +348,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
@param request: The subscribe request containing node identifier, subscriber JID,
|
||||
and optional subscription options.
|
||||
@return: A Subscription object representing the resulting subscription state.
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@raise StanzaError: with "item-not-found" if the node does not exist
|
||||
@raise StanzaError: with "forbidden" if the requestor lacks subscribe privileges
|
||||
@raise StanzaError: with "not-authorized" if there's a pending subscription
|
||||
"""
|
||||
|
|
@ -337,7 +370,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
raise error.StanzaError("item-not-found", text="Node does not exist")
|
||||
|
||||
affiliation = next(
|
||||
(
|
||||
|
|
@ -407,7 +440,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
"""Handle an unsubscribe request.
|
||||
|
||||
@param request: The unsubscribe request
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@raise StanzaError: with "item-not-found" if the node does not exist
|
||||
@raise StanzaError: with "forbidden" if the requestor is not the subscriber
|
||||
"""
|
||||
client = self.client
|
||||
|
|
@ -427,7 +460,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
node = await G.storage.get_pubsub_node(client, client.jid, request.nodeIdentifier)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
raise error.StanzaError("item-not-found", text="Node does not exist")
|
||||
|
||||
await G.storage.set_pubsub_subscriptions(
|
||||
node,
|
||||
|
|
@ -601,7 +634,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
|
||||
@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 "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
|
||||
|
|
@ -615,7 +648,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
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):
|
||||
|
|
@ -660,9 +693,9 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
async def _configureSet(self, request: pubsub.PubSubRequest) -> None:
|
||||
"""Set the configuration of a node.
|
||||
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@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
|
||||
@raise StanzaError: with "not-acceptable" if configuration values are invalid
|
||||
"""
|
||||
client = self.client
|
||||
assert request.sender is not None
|
||||
|
|
@ -675,7 +708,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
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):
|
||||
|
|
@ -742,18 +775,20 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
def items(
|
||||
self,
|
||||
request: pubsub.PubSubRequest,
|
||||
) -> defer.Deferred[tuple[list[domish.Element], rsm.RSMResponse | None]]:
|
||||
) -> defer.Deferred[
|
||||
list[domish.Element] | tuple[list[domish.Element], rsm.RSMResponse]
|
||||
]:
|
||||
return defer.ensureDeferred(self._items(request))
|
||||
|
||||
async def _items(
|
||||
self, request: pubsub.PubSubRequest
|
||||
) -> tuple[list[domish.Element], rsm.RSMResponse | None]:
|
||||
) -> list[domish.Element] | tuple[list[domish.Element], rsm.RSMResponse]:
|
||||
"""Retrieve items from a pubsub node.
|
||||
|
||||
@param request: The items request containing node identifier, item identifiers,
|
||||
max items limit, and optional RSM parameters.
|
||||
@return: A tuple of (list of item elements, RSM response or None).
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@raise StanzaError: with "item-not-found" if the node does not exist
|
||||
@raise StanzaError: with "forbidden" if the requestor lacks retrieve privileges
|
||||
"""
|
||||
client = self.client
|
||||
|
|
@ -767,7 +802,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
raise error.StanzaError("item-not-found", text="Node does not exist")
|
||||
|
||||
affiliation = next(
|
||||
(
|
||||
|
|
@ -821,19 +856,164 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
return defer.ensureDeferred(self._retract(request))
|
||||
|
||||
async def _retract(self, request: pubsub.PubSubRequest) -> None:
|
||||
return
|
||||
"""Handle items retraction request.
|
||||
|
||||
@param request: The retract request
|
||||
@raise StanzaError: with "item-not-found" if the node does not exist
|
||||
@raise StanzaError: with "forbidden" if the requestor lacks retract privileges
|
||||
@raise StanzaError: with "bad-request" if no item identifiers are provided
|
||||
"""
|
||||
client = self.client
|
||||
assert request.sender is not None
|
||||
requestor_jid = request.sender.userhostJID()
|
||||
|
||||
assert request.nodeIdentifier is not None
|
||||
|
||||
if not request.itemIdentifiers:
|
||||
raise error.StanzaError("bad-request", text="No item identifiers provided")
|
||||
|
||||
node = await G.storage.get_pubsub_node(
|
||||
client,
|
||||
client.jid,
|
||||
request.nodeIdentifier,
|
||||
with_affiliations=True,
|
||||
with_subscriptions=True,
|
||||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item-not-found", text="Node does not exist")
|
||||
|
||||
items_data, __ = await G.storage.get_items(
|
||||
node=node,
|
||||
max_items=None,
|
||||
item_ids=request.itemIdentifiers,
|
||||
)
|
||||
|
||||
existing_items = [
|
||||
item for item in items_data if item.name in request.itemIdentifiers
|
||||
]
|
||||
|
||||
if not existing_items:
|
||||
raise error.StanzaError(
|
||||
"item-not-found",
|
||||
text="No items found with the specified identifiers",
|
||||
)
|
||||
|
||||
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 not in (Affiliation.owner, Affiliation.publisher):
|
||||
for item in existing_items:
|
||||
publisher_jid = item.data.getAttribute("publisher")
|
||||
if (
|
||||
not publisher_jid
|
||||
or jid.JID(publisher_jid).userhostJID() != requestor_jid
|
||||
):
|
||||
raise error.StanzaError(
|
||||
"forbidden",
|
||||
text="Insufficient privileges to retract this item",
|
||||
)
|
||||
|
||||
await G.storage.delete_pubsub_items(node, request.itemIdentifiers)
|
||||
|
||||
if request.notify:
|
||||
item_elts = [item.data for item in existing_items]
|
||||
await self._send_notifications(node, "retract", item_elts)
|
||||
|
||||
def purge(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]:
|
||||
return defer.ensureDeferred(self._purge(request))
|
||||
|
||||
async def _purge(self, request: pubsub.PubSubRequest) -> None:
|
||||
return
|
||||
"""Handle a node purge request.
|
||||
|
||||
@param request: The purge request.
|
||||
@raise StanzaError: with "item-not-found" if the node does not exist
|
||||
@raise StanzaError: with "forbidden" if the requestor is not an owner
|
||||
"""
|
||||
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,
|
||||
with_subscriptions=True,
|
||||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item-not-found", text="Node does not exist")
|
||||
|
||||
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")
|
||||
|
||||
await G.storage.delete_pubsub_items(node, items_names=None)
|
||||
await self._send_notifications(node, "purge", [])
|
||||
|
||||
def delete(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]:
|
||||
return defer.ensureDeferred(self._delete(request))
|
||||
|
||||
async def _delete(self, request: pubsub.PubSubRequest) -> None:
|
||||
return
|
||||
"""Handle a node deletion request.
|
||||
|
||||
@param request: The delete request containing the node identifier.
|
||||
@raise StanzaError: with "item-not-found" if the node does not exist
|
||||
@raise StanzaError: with "forbidden" if the requestor is not an owner
|
||||
"""
|
||||
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,
|
||||
with_subscriptions=True,
|
||||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item-not-found", text="Node does not exist")
|
||||
|
||||
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")
|
||||
|
||||
await self._send_notifications(node, "delete", [])
|
||||
await G.storage.delete_pubsub_node(
|
||||
profiles=None, services=[client.jid], names=[node.name]
|
||||
)
|
||||
|
||||
def affiliationsGet(
|
||||
self, request: pubsub.PubSubRequest
|
||||
|
|
@ -845,7 +1025,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
|
||||
@param request: The affiliations get request with node identifier.
|
||||
@return: Mapping of entity JIDs to their affiliations.
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@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
|
||||
|
|
@ -859,7 +1039,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
raise error.StanzaError("item-not-found", text="Node does not exist")
|
||||
|
||||
if not G.host.memory.is_admin_jid(requestor_jid):
|
||||
affiliation = next(
|
||||
|
|
@ -887,9 +1067,9 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
|
||||
@param request: The affiliations set request with node identifier and affiliations
|
||||
dict.
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@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 affiliation values are invalid
|
||||
@raise StanzaError: with "not-acceptable" if affiliation values are invalid
|
||||
"""
|
||||
client = self.client
|
||||
assert request.sender is not None
|
||||
|
|
@ -902,7 +1082,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
raise error.StanzaError("item-not-found", text="Node does not exist")
|
||||
|
||||
if not G.host.memory.is_admin_jid(requestor_jid):
|
||||
affiliation = next(
|
||||
|
|
@ -960,7 +1140,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
|
||||
@param request: The subscriptions get request with node identifier.
|
||||
@return: List of Subscription objects for all subscribers of the node.
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@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
|
||||
|
|
@ -978,7 +1158,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
raise error.StanzaError("item-not-found", text="Node does not exist")
|
||||
|
||||
if not G.host.memory.is_admin_jid(requestor_jid):
|
||||
affiliation = next(
|
||||
|
|
@ -1020,9 +1200,9 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
|
||||
@param request: The subscriptions set request with node identifier and
|
||||
subscriptions dict.
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@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 subscription values are invalid
|
||||
@raise StanzaError: with "not-acceptable" if subscription values are invalid
|
||||
"""
|
||||
client = self.client
|
||||
assert request.sender is not None
|
||||
|
|
@ -1035,7 +1215,7 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
raise error.StanzaError("item-not-found", text="Node does not exist")
|
||||
|
||||
if not G.host.memory.is_admin_jid(requestor_jid):
|
||||
affiliation = next(
|
||||
|
|
@ -1180,7 +1360,10 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
for item_elt in item_elts:
|
||||
retract_elt = event_type_elt.addElement("retract")
|
||||
retract_elt["id"] = item_elt.getAttribute("id")
|
||||
case "delete" | "purge" | "configuration":
|
||||
case "delete" | "purge":
|
||||
# Empty elements, no children needed
|
||||
pass
|
||||
case "configuration":
|
||||
raise NotImplementedError
|
||||
case _:
|
||||
raise exceptions.InternalError(
|
||||
|
|
|
|||
Loading…
Reference in a new issue