libervia-backend: libervia/backend/plugins/plugin_comp_ap_gateway/pubsub

comparison libervia/backend/plugins/plugin_comp_ap_gateway/pubsub_service.py @ 4270:0d7bb4df2343

Reformatted code base using black.

author	Goffi <goffi@goffi.org>
date	Wed, 19 Jun 2024 18:44:57 +0200
parents	49019947cc76
children

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 from libervia.backend.tools import image
 from libervia.backend.tools.utils import ensure_deferred
 from libervia.backend.tools.web import download_file
 from libervia.backend.memory.sqla_mapping import PubsubSub, SubscriptionState
-from .constants import (
+from .constants import TYPE_ACTOR, ST_AVATAR, MAX_AVATAR_SIZE
-TYPE_ACTOR,
-ST_AVATAR,
-MAX_AVATAR_SIZE
-)
 log = getLogger(__name__)
 # all nodes have the same config
 NODE_CONFIG = [
 {"var": "pubsub#persist_items", "type": "boolean", "value": True},
 {"var": "pubsub#max_items", "value": "max"},
 {"var": "pubsub#access_model", "type": "list-single", "value": "open"},
 {"var": "pubsub#publish_model", "type": "list-single", "value": "open"},
 ]
 NODE_CONFIG_VALUES = {c["var"]: c["value"] for c in NODE_CONFIG}
 NODE_OPTIONS = {c["var"]: {} for c in NODE_CONFIG}
 for c in NODE_CONFIG:
-NODE_OPTIONS[c["var"]].update({k:v for k,v in c.items() if k not in ("var", "value")})
+NODE_OPTIONS[c["var"]].update(
+{k: v for k, v in c.items() if k not in ("var", "value")}
+)
 class APPubsubService(rsm.PubSubService):
 """Pubsub service for XMPP requests"""
 @return: requestor actor ID, recipient actor ID and recipient inbox
 @raise error.StanzaError: "item-not-found" is raised if not user part is specified
 in requestor
 """
 if not recipient.user:
-raise error.StanzaError(
+raise error.StanzaError("item-not-found", text="No user part specified")
-"item-not-found",
-text="No user part specified"
-)
 requestor_actor_id = self.apg.build_apurl(TYPE_ACTOR, requestor.userhost())
 recipient_account = self.apg._e.unescape(recipient.user)
-recipient_actor_id = await self.apg.get_ap_actor_id_from_account(recipient_account)
+recipient_actor_id = await self.apg.get_ap_actor_id_from_account(
+recipient_account
+)
 inbox = await self.apg.get_ap_inbox_from_id(recipient_actor_id, use_shared=False)
 return requestor_actor_id, recipient_actor_id, inbox
 @ensure_deferred
 async def publish(self, requestor, service, nodeIdentifier, items):
 if self.apg.local_only and not self.apg.is_local(requestor):
 raise error.StanzaError(
-"forbidden",
+"forbidden", "Only local users can publish on this gateway."
-"Only local users can publish on this gateway."
 )
 if not service.user:
 raise error.StanzaError(
 "bad-request",
-"You must specify an ActivityPub actor account in JID user part."
+"You must specify an ActivityPub actor account in JID user part.",
 )
 ap_account = self.apg._e.unescape(service.user)
 if ap_account.count("@") != 1:
 raise error.StanzaError(
-"bad-request",
+"bad-request", f"{ap_account!r} is not a valid ActivityPub actor account."
-f"{ap_account!r} is not a valid ActivityPub actor account."
 )
 client = self.apg.client.get_virtual_client(requestor)
 if self.apg._pa.is_attachment_node(nodeIdentifier):
 await self.apg.convert_and_post_attachments(
 client, ap_account, service, nodeIdentifier, items
 )
 cached_node = await self.host.memory.storage.get_pubsub_node(
 client, service, nodeIdentifier, with_subscriptions=True, create=True
 )
-await self.host.memory.storage.cache_pubsub_items(
+await self.host.memory.storage.cache_pubsub_items(client, cached_node, items)
-client,
-cached_node,
-items
-)
 for subscription in cached_node.subscriptions:
 if subscription.state != SubscriptionState.SUBSCRIBED:
 continue
 self.notifyPublish(
-service,
+service, nodeIdentifier, [(subscription.subscriber, None, items)]
-nodeIdentifier,
+)
-[(subscription.subscriber, None, items)]
-)
+async def ap_following_2_elt(
+self, requestor_actor_id: str, ap_item: dict
-async def ap_following_2_elt(self, requestor_actor_id: str, ap_item: dict) -> domish.Element:
+) -> domish.Element:
 """Convert actor ID from following collection to XMPP item
 @param requestor_actor_id: ID of the actor doing the request.
 @param ap_item: AP item from which actor ID must be extracted.
 """
 )
 item_elt = pubsub.Item(id=actor_id, payload=subscription_elt)
 return item_elt
 async def ap_follower_2_elt(
-self,
+self, requestor_actor_id: str, ap_item: dict
-requestor_actor_id: str,
-ap_item: dict
 ) -> domish.Element:
 """Convert actor ID from followers collection to XMPP item
 @param requestor_actor_id: ID of the actor doing the request.
 @param ap_item: AP item from which actor ID must be extracted.
 subscriber_elt = self.apg._pps.build_subscriber_elt(actor_jid)
 item_elt = pubsub.Item(id=actor_id, payload=subscriber_elt)
 return item_elt
 async def generate_v_card(
-self,
+self, requestor_actor_id: str, ap_account: str
-requestor_actor_id: str,
-ap_account: str
 ) -> domish.Element:
 """Generate vCard4 (XEP-0292) item element from ap_account's metadata
 @param requestor_actor_id: ID of the actor doing the request.
 @param ap_account: AP account from where the vcard must be retrieved.
 @return: <item> with the <vcard> element
 """
 actor_data = await self.apg.get_ap_actor_data_from_account(
-requestor_actor_id,
+requestor_actor_id, ap_account
-ap_account
 )
 identity_data = {}
 summary = actor_data.get("summary")
 # summary is HTML, we have to convert it to text
 item_elt.addChild(vcard_elt)
 item_elt["id"] = self.apg._p.ID_SINGLETON
 return item_elt
 async def get_avatar_data(
-self,
+self, client: SatXMPPEntity, requestor_actor_id: str, ap_account: str
-client: SatXMPPEntity,
-requestor_actor_id: str,
-ap_account: str
 ) -> dict[str, Any]:
 """Retrieve actor's avatar if any, cache it and file actor_data
 @param client: client to use for the request.
 @param requestor_actor_id: ID of the actor doing the request.
 dest_path = Path(dir_name, filename)
 await download_file(url, dest_path, max_size=MAX_AVATAR_SIZE)
 avatar_data = {
 "path": dest_path,
 "filename": filename,
-'media_type': image.guess_type(dest_path),
+"media_type": image.guess_type(dest_path),
 }
-await self.apg._i.cache_avatar(
+await self.apg._i.cache_avatar(self.apg.IMPORT_NAME, avatar_data)
-self.apg.IMPORT_NAME,
-avatar_data
-)
 else:
 avatar_data = {
 "cache_uid": cache["uid"],
 "path": cache["path"],
-"media_type": cache["mime_type"]
+"media_type": cache["mime_type"],
 }
 return avatar_data
 async def generate_avatar_metadata(
-self,
+self, client: SatXMPPEntity, requestor_actor_id: str, ap_account: str
-client: SatXMPPEntity,
-requestor_actor_id: str,
-ap_account: str
 ) -> domish.Element:
 """Generate the metadata element for user avatar
 @param requestor_actor_id: ID of the actor doing the request.
 @raise StanzaError("item-not-found"): no avatar is present in actor data (in
 @param requestor_actor_id: ID of the actor doing the request.
 @raise StanzaError("item-not-found"): no avatar cached with requested ID
 """
 if not itemIdentifiers:
 avatar_data = await self.get_avatar_data(
-client,
+client, requestor_actor_id, ap_account
-requestor_actor_id,
-ap_account
 )
 if "base64" not in avatar_data:
 await threads.deferToThread(
-self._blocking_b_6_4_encode_avatar,
+self._blocking_b_6_4_encode_avatar, avatar_data
-avatar_data
 )
 else:
 if len(itemIdentifiers) > 1:
 # only a single item ID is supported
 raise error.StanzaError("item-not-found")
 # just to be sure that that we don't have an empty string
 assert item_id
 cache_data = self.apg.host.common_cache.get_metadata(item_id)
 if cache_data is None:
 raise error.StanzaError("item-not-found")
-avatar_data = {
+avatar_data = {"cache_uid": item_id, "path": cache_data["path"]}
-"cache_uid": item_id,
-"path": cache_data["path"]
-}
 await threads.deferToThread(self._blocking_b_6_4_encode_avatar, avatar_data)
 return self.apg._a.build_item_data_elt(avatar_data)
 @ensure_deferred
 requestor: jid.JID,
 service: jid.JID,
 node: str,
 maxItems: Optional[int],
 itemIdentifiers: Optional[List[str]],
-rsm_req: Optional[rsm.RSMRequest]
+rsm_req: Optional[rsm.RSMRequest],
 ) -> Tuple[List[domish.Element], Optional[rsm.RSMResponse]]:
 if not service.user:
 return [], None
 ap_account = self.host.plugins["XEP-0106"].unescape(service.user)
 if ap_account.count("@") != 1:
 log.warning(f"Invalid AP account used by {requestor}: {ap_account!r}")
 return [], None
 requestor_actor_id = self.apg.build_apurl(
-TYPE_ACTOR,
+TYPE_ACTOR, await self.apg.get_ap_account_from_jid_and_node(service, node)
-await self.apg.get_ap_account_from_jid_and_node(service, node)
 )
 # cached_node may be pre-filled with some nodes (e.g. attachments nodes),
 # otherwise it is filled when suitable
 cached_node = None
 # vCard4 request
 item_elt = await self.generate_v_card(requestor_actor_id, ap_account)
 return [item_elt], None
 elif node == self.apg._a.namespace_metadata:
 item_elt = await self.generate_avatar_metadata(
-self.apg.client,
+self.apg.client, requestor_actor_id, ap_account
-requestor_actor_id,
-ap_account
 )
 return [item_elt], None
 elif node == self.apg._a.namespace_data:
 item_elt = await self.generate_avatar_data(
-self.apg.client,
+self.apg.client, requestor_actor_id, ap_account, itemIdentifiers
-requestor_actor_id,
-ap_account,
-itemIdentifiers
 )
 return [item_elt], None
 elif self.apg._pa.is_attachment_node(node):
 use_cache = True
 # we check cache here because we emit an item-not-found error if the node is
 parser = self.apg.ap_events.ap_item_2_event_elt
 else:
 raise error.StanzaError(
 "feature-not-implemented",
 text=f"AP Gateway {C.APP_VERSION} only supports "
-f"{self.apg._m.namespace} node for now"
+f"{self.apg._m.namespace} node for now",
 )
 collection_name = "outbox"
 use_cache = True
 if use_cache:
 return items, None
 else:
 if rsm_req is None:
 if maxItems is None:
 maxItems = 20
-kwargs.update({
+kwargs.update(
-"max_items": maxItems,
+{
-"chronological_pagination": False,
+"max_items": maxItems,
-})
+"chronological_pagination": False,
+}
+)
 else:
-if len(
+if (
-[v for v in (rsm_req.after, rsm_req.before, rsm_req.index)
+len(
-if v is not None]
+[
-) > 1:
+v
+for v in (rsm_req.after, rsm_req.before, rsm_req.index)
+if v is not None
+]
+)
+> 1
+):
 raise error.StanzaError(
 "bad-request",
-text="You can't use after, before and index at the same time"
+text="You can't use after, before and index at the same time",
 )
 kwargs.update({"max_items": rsm_req.max})
 if rsm_req.after is not None:
 kwargs["after_id"] = rsm_req.after
 elif rsm_req.before is not None:
 if self.apg._m.is_comment_node(node):
 parent_item = self.apg._m.get_parent_item(node)
 try:
 parent_data = await self.apg.ap_get(parent_item, requestor_actor_id)
 collection = await self.apg.ap_get_object(
-requestor_actor_id,
+requestor_actor_id, parent_data.get("object", {}), "replies"
-parent_data.get("object", {}),
-"replies"
 )
 except Exception as e:
-raise error.StanzaError(
+raise error.StanzaError("item-not-found", text=str(e))
-"item-not-found",
-text=str(e)
-)
 else:
 actor_data = await self.apg.get_ap_actor_data_from_account(
-requestor_actor_id,
+requestor_actor_id, ap_account
-ap_account
+)
-)
+collection = await self.apg.ap_get_object(
-collection = await self.apg.ap_get_object(requestor_actor_id, actor_data, collection_name)
+requestor_actor_id, actor_data, collection_name
+)
 if not collection:
 raise error.StanzaError(
 "item-not-found",
-text=f"No collection found for node {node!r} (account: {ap_account})"
+text=f"No collection found for node {node!r} (account: {ap_account})",
 )
 kwargs["parser"] = parser
 return await self.apg.get_ap_items(requestor_actor_id, collection, **kwargs)
 )
 subscription = None
 else:
 try:
 subscription = next(
-s for s in node.subscriptions
+s
+for s in node.subscriptions
 if s.subscriber == requestor.userhostJID()
 )
 except StopIteration:
 subscription = None
 if subscription is None:
-subscription = PubsubSub(
+subscription = PubsubSub(subscriber=requestor.userhostJID(), state=sub_state)
-subscriber=requestor.userhostJID(),
-state=sub_state
-)
 node.subscriptions.append(subscription)
 await self.host.memory.storage.add(node)
 else:
 if subscription.state is None:
 subscription.state = sub_state
 requestor, service
 )
 data = self.apg.create_activity(
 "Undo",
 req_actor_id,
-self.apg.create_activity(
+self.apg.create_activity("Follow", req_actor_id, recip_actor_id),
-"Follow",
-req_actor_id,
-recip_actor_id
-)
 )
 resp = await self.apg.sign_and_post(inbox, req_actor_id, data)
 if resp.code >= 300:
 text = await resp.text()
 def getConfigurationOptions(self):
 return NODE_OPTIONS
 def getConfiguration(
-self,
+self, requestor: jid.JID, service: jid.JID, nodeIdentifier: str
-requestor: jid.JID,
-service: jid.JID,
-nodeIdentifier: str
 ) -> defer.Deferred:
 return defer.succeed(NODE_CONFIG_VALUES)
 def getNodeInfo(
 self,
 requestor: jid.JID,
 service: jid.JID,
 nodeIdentifier: str,
 pep: bool = False,
-recipient: Optional[jid.JID] = None
+recipient: Optional[jid.JID] = None,
 ) -> Optional[dict]:
 if not nodeIdentifier:
 return None
-info = {
+info = {"type": "leaf", "meta-data": NODE_CONFIG}
-"type": "leaf",
-"meta-data": NODE_CONFIG
-}
 return info

Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_comp_ap_gateway/pubsub_service.py @ 4270:0d7bb4df2343