Mercurial > libervia-backend
diff sat/plugins/plugin_comp_ap_gateway/__init__.py @ 3742:bf0505d41c09
comp AP: helper methods to get AP objects:
rel 364
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 22 Mar 2022 17:00:42 +0100 |
parents | 86eea17cafa7 |
children | a8c7e5cef0cb |
line wrap: on
line diff
--- a/sat/plugins/plugin_comp_ap_gateway/__init__.py Tue Mar 22 17:00:42 2022 +0100 +++ b/sat/plugins/plugin_comp_ap_gateway/__init__.py Tue Mar 22 17:00:42 2022 +0100 @@ -211,6 +211,135 @@ text=f"Can't get AP data at {url}: {e}" ) + @overload + async def apGetObject(self, data: dict, key: str) -> Optional[dict]: + ... + + @overload + async def apGetObject( + self, data: Union[str, dict], key: None = None + ) -> dict: + ... + + async def apGetObject(self, data, key = None): + """Retrieve an AP object, dereferencing when necessary + + This method is to be used with attributes marked as "Functional" in + https://www.w3.org/TR/activitystreams-vocabulary + @param data: AP object where an other object is looked for, or the object itself + @param key: name of the object to look for, or None if data is the object directly + @return: found object if any + """ + if key is not None: + value = data.get(key) + else: + value = data + if value is None: + if key is None: + raise ValueError("None can't be used with apGetObject is key is None") + return None + elif isinstance(value, dict): + return value + elif isinstance(value, str): + return await self.apGet(value) + else: + raise NotImplementedError( + "was expecting a string or a dict, got {type(value)}: {value!r}}" + ) + + async def apGetList(self, data: dict, key: str) -> Optional[List[dict]]: + """Retrieve a list of objects from AP data, dereferencing when necessary + + This method is to be used with non functional vocabularies. Use ``apGetObject`` + otherwise. + If the value is a dictionary, it will be wrapped in a list + @param data: AP object where a list of objects is looked for + @param key: key of the list to look for + @return: list of objects, or None if the key is not present + """ + value = data.get(key) + if value is None: + return None + elif isinstance(value, str): + value = await self.apGet(value) + if isinstance(value, dict): + return [value] + if not isinstance(value, list): + raise ValueError(f"A list was expected, got {type(value)}: {value!r}") + return [await self.apGetObject(i) for i in value] + + async def apGetActors( + self, + data: dict, + key: str, + as_account: bool = True + ) -> List[str]: + """Retrieve AP actors from data + + @param data: AP object containing a field with actors + @param key: field to use to retrieve actors + @param as_account: if True returns account handles, otherwise will return actor + IDs + @raise exceptions.DataError: there is not actor data or it is invalid + """ + value = data.get(key) + if value is None: + raise exceptions.DataError( + f"no actor associated to object {data.get('id')!r}" + ) + elif isinstance(value, dict): + actor_id = value.get("id") + if actor_id is None: + raise exceptions.DataError( + f"invalid actor associated to object {data.get('id')!r}: {value!r}" + ) + value = [actor_id] + elif isinstance(value, str): + value = [value] + elif isinstance(value, list): + try: + value = [a if isinstance(a, str) else a["id"] for a in value] + except (TypeError, KeyError): + raise exceptions.DataError( + f"invalid actors list to object {data.get('id')!r}: {value!r}" + ) + if not value: + raise exceptions.DataError( + f"list of actors is empty" + ) + if as_account: + return [await self.getAPAccountFromId(actor_id) for actor_id in value] + else: + return value + + async def apGetSenderActor( + self, + data: dict, + ) -> str: + """Retrieve actor who sent data + + This is done by checking "attributedTo" field first, then "actor" field. + Only the first found actor is taken into accoun + @param data: AP object + @return: actor id of the sender + @raise exceptions.NotFound: no actor has been found in data + """ + try: + actors = await self.apGetActors(data, "attributedTo", as_account=False) + except exceptions.DataError: + actors = None + if not actors: + try: + actors = await self.apGetActors(data, "actor", as_account=False) + except exceptions.DataError: + raise exceptions.NotFound( + 'actor not specified in "attributedTo" or "actor"' + ) + try: + return actors[0] + except IndexError: + raise exceptions.NotFound("list of actors is empty") + def mustEncode(self, text: str) -> bool: """Indicate if a text must be period encoded""" return (