Mercurial > libervia-backend

comparison sat/plugins/plugin_comp_ap_gateway/__init__.py @ 3742:bf0505d41c09

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
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
comparison
equal deleted inserted replaced
3741:eddab3798aca 3742:bf0505d41c09
208 except Exception as e: 208 except Exception as e:
209 raise error.StanzaError( 209 raise error.StanzaError(
210 "service-unavailable", 210 "service-unavailable",
211 text=f"Can't get AP data at {url}: {e}" 211 text=f"Can't get AP data at {url}: {e}"
212 ) 212 )
213
214 @overload
215 async def apGetObject(self, data: dict, key: str) -> Optional[dict]:
216 ...
217
218 @overload
219 async def apGetObject(
220 self, data: Union[str, dict], key: None = None
221 ) -> dict:
222 ...
223
224 async def apGetObject(self, data, key = None):
225 """Retrieve an AP object, dereferencing when necessary
226
227 This method is to be used with attributes marked as "Functional" in
228 https://www.w3.org/TR/activitystreams-vocabulary
229 @param data: AP object where an other object is looked for, or the object itself
230 @param key: name of the object to look for, or None if data is the object directly
231 @return: found object if any
232 """
233 if key is not None:
234 value = data.get(key)
235 else:
236 value = data
237 if value is None:
238 if key is None:
239 raise ValueError("None can't be used with apGetObject is key is None")
240 return None
241 elif isinstance(value, dict):
242 return value
243 elif isinstance(value, str):
244 return await self.apGet(value)
245 else:
246 raise NotImplementedError(
247 "was expecting a string or a dict, got {type(value)}: {value!r}}"
248 )
249
250 async def apGetList(self, data: dict, key: str) -> Optional[List[dict]]:
251 """Retrieve a list of objects from AP data, dereferencing when necessary
252
253 This method is to be used with non functional vocabularies. Use ``apGetObject``
254 otherwise.
255 If the value is a dictionary, it will be wrapped in a list
256 @param data: AP object where a list of objects is looked for
257 @param key: key of the list to look for
258 @return: list of objects, or None if the key is not present
259 """
260 value = data.get(key)
261 if value is None:
262 return None
263 elif isinstance(value, str):
264 value = await self.apGet(value)
265 if isinstance(value, dict):
266 return [value]
267 if not isinstance(value, list):
268 raise ValueError(f"A list was expected, got {type(value)}: {value!r}")
269 return [await self.apGetObject(i) for i in value]
270
271 async def apGetActors(
272 self,
273 data: dict,
274 key: str,
275 as_account: bool = True
276 ) -> List[str]:
277 """Retrieve AP actors from data
278
279 @param data: AP object containing a field with actors
280 @param key: field to use to retrieve actors
281 @param as_account: if True returns account handles, otherwise will return actor
282 IDs
283 @raise exceptions.DataError: there is not actor data or it is invalid
284 """
285 value = data.get(key)
286 if value is None:
287 raise exceptions.DataError(
288 f"no actor associated to object {data.get('id')!r}"
289 )
290 elif isinstance(value, dict):
291 actor_id = value.get("id")
292 if actor_id is None:
293 raise exceptions.DataError(
294 f"invalid actor associated to object {data.get('id')!r}: {value!r}"
295 )
296 value = [actor_id]
297 elif isinstance(value, str):
298 value = [value]
299 elif isinstance(value, list):
300 try:
301 value = [a if isinstance(a, str) else a["id"] for a in value]
302 except (TypeError, KeyError):
303 raise exceptions.DataError(
304 f"invalid actors list to object {data.get('id')!r}: {value!r}"
305 )
306 if not value:
307 raise exceptions.DataError(
308 f"list of actors is empty"
309 )
310 if as_account:
311 return [await self.getAPAccountFromId(actor_id) for actor_id in value]
312 else:
313 return value
314
315 async def apGetSenderActor(
316 self,
317 data: dict,
318 ) -> str:
319 """Retrieve actor who sent data
320
321 This is done by checking "attributedTo" field first, then "actor" field.
322 Only the first found actor is taken into accoun
323 @param data: AP object
324 @return: actor id of the sender
325 @raise exceptions.NotFound: no actor has been found in data
326 """
327 try:
328 actors = await self.apGetActors(data, "attributedTo", as_account=False)
329 except exceptions.DataError:
330 actors = None
331 if not actors:
332 try:
333 actors = await self.apGetActors(data, "actor", as_account=False)
334 except exceptions.DataError:
335 raise exceptions.NotFound(
336 'actor not specified in "attributedTo" or "actor"'
337 )
338 try:
339 return actors[0]
340 except IndexError:
341 raise exceptions.NotFound("list of actors is empty")
213 342
214 def mustEncode(self, text: str) -> bool: 343 def mustEncode(self, text: str) -> bool:
215 """Indicate if a text must be period encoded""" 344 """Indicate if a text must be period encoded"""
216 return ( 345 return (
217 not RE_ALLOWED_UNQUOTED.match(text) 346 not RE_ALLOWED_UNQUOTED.match(text)