add ap object classes
This commit is contained in:
parent
0442c7b66e
commit
443b949ea5
290
aputils/message.py
Normal file
290
aputils/message.py
Normal file
|
@ -0,0 +1,290 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from typing import Any, Callable, Optional, Type
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from .enums import ObjectType
|
||||
from .signer import Signer
|
||||
|
||||
|
||||
OBJECT_TYPES = {}
|
||||
|
||||
|
||||
def register_object_type(*types: ObjectType) -> Callable[[Type[Object]], Type[Object]]:
|
||||
"""
|
||||
Decorator to set an ``Object``-based class to an ActivityPub object type for parsing
|
||||
|
||||
:param types:
|
||||
"""
|
||||
def wrapper(cls: Type[Object]) -> Type[Object]:
|
||||
for obj_type in types:
|
||||
OBJECT_TYPES[ObjectType.parse(obj_type)] = cls
|
||||
|
||||
return cls
|
||||
return wrapper
|
||||
|
||||
|
||||
class Object(dict):
|
||||
":class:`dict` object that represents an ActivityPub Object"
|
||||
|
||||
|
||||
@classmethod
|
||||
def new(cls: Type[Object], obj_type: ObjectType, context: list[Any] = None,
|
||||
**kwargs: Any) -> Object:
|
||||
"""
|
||||
Create a new ActivityPub object
|
||||
|
||||
:param obj_type:
|
||||
"""
|
||||
obj_type = ObjectType.parse(obj_type)
|
||||
|
||||
if not context:
|
||||
context = []
|
||||
|
||||
elif not isinstance(context, (list, tuple, set)):
|
||||
raise TypeError("Object context must be a list, tuple, or set")
|
||||
|
||||
if "https://www.w3.org/ns/activitystreams" not in context:
|
||||
context.insert(0, "https://www.w3.org/ns/activitystreams")
|
||||
|
||||
try:
|
||||
OBJECT_TYPES[obj_type]({"@context": [context]}, type = obj_type, **kwargs)
|
||||
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
return cls({"@context": [context]}, type = obj_type, **kwargs)
|
||||
|
||||
|
||||
@classmethod
|
||||
def parse(cls: Type[Object], data: str | bytes | dict | Object) -> Object:
|
||||
"""
|
||||
Parse an ActivityPub object and return the appropriate ``Object`` object
|
||||
|
||||
:param data: ActivityPub object as a :class:`str`, :class:`bytes`, :class:`dict`, or
|
||||
:class:`Object` object
|
||||
"""
|
||||
if isinstance(data, (str, bytes)):
|
||||
data = json.loads(data)
|
||||
|
||||
if not isinstance(data, dict):
|
||||
raise TypeError(f"Cannot parse objects of type \"{type(data).__name__}\"")
|
||||
|
||||
try:
|
||||
return OBJECT_TYPES[data["type"]](data)
|
||||
|
||||
except KeyError:
|
||||
return Object(data)
|
||||
|
||||
|
||||
@property
|
||||
def domain(self) -> str:
|
||||
"Get the domain of the object origin"
|
||||
return urlparse(self["id"]).hostname
|
||||
|
||||
|
||||
@property
|
||||
def object_id(self) -> str:
|
||||
"Get the ``id`` field of a linked object if it exists"
|
||||
try:
|
||||
return self["object"]["id"]
|
||||
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
return self["object"]
|
||||
|
||||
|
||||
@property
|
||||
def object_domain(self) -> str:
|
||||
"Get the domain of the linked object if it exists"
|
||||
return urlparse(self.object_id).hostname
|
||||
|
||||
|
||||
@register_object_type(*ObjectType.Activity)
|
||||
class Activity(Object):
|
||||
":class:`dict` object that reepresents an ActivityPub activity"
|
||||
|
||||
|
||||
@register_object_type(*ObjectType.Actor)
|
||||
class Actor(Object):
|
||||
":class:`dict` object that represents an ActivityPub actor"
|
||||
|
||||
_signer: Signer = None
|
||||
|
||||
|
||||
def __setitem__(self, key: str, value) -> None: # noqa: ANN001
|
||||
if key == "attachment":
|
||||
value = [Attachment(item) for item in value] if value else []
|
||||
|
||||
Object.__setitem__(self, key, value)
|
||||
|
||||
|
||||
@classmethod
|
||||
def new(cls: Type[Actor], actor_type: ObjectType, handle: str, actor: str, pubkey: str,
|
||||
inbox: Optional[str] = None, outbox: Optional[str] = None,
|
||||
shared_inbox: Optional[str] = None, fields: Optional[dict[str, str]] = None,
|
||||
avatar: Optional[str] = None, header: Optional[str] = None, **kwargs: Any) -> Actor:
|
||||
"""
|
||||
Create a new actor object
|
||||
|
||||
:param actor_type: Type of actor to create
|
||||
:param handle: Username of the actor
|
||||
:param actor: URL to where this object will be hosted
|
||||
:param pubkey: PEM of the private key associated with the actor
|
||||
:param inbox: URL to the actor's inbox
|
||||
:param outbox: URL to the actor's outbox,
|
||||
:param shared_inbox: URL to the inbox shared amongst all users of the instance
|
||||
:param fields: key/value string pairs to display on the actor's profile page
|
||||
:param avatar: URL to an image to be used as the actor's profile picture
|
||||
:param header: URL to an image to be used as the actor's header image
|
||||
:param kwargs: Extra object values to set
|
||||
"""
|
||||
|
||||
actor_type = ObjectType.parse(type)
|
||||
|
||||
if actor_type not in ObjectType.Actor:
|
||||
raise ValueError(f"Invalid Actor type: {actor_type.value}")
|
||||
|
||||
parsed_url = urlparse(actor)
|
||||
proto = parsed_url.scheme
|
||||
domain = parsed_url.netloc
|
||||
|
||||
data = Object.new(
|
||||
actor_type,
|
||||
context = [
|
||||
{
|
||||
"toot": "http://joinmastodon.org/ns#",
|
||||
"publicKeyBase64": "toot:publicKeyBase64",
|
||||
"schema": "http://schema.org#",
|
||||
"PropertyValue": "schema:PropertyValue",
|
||||
"value": "schema:value"
|
||||
}
|
||||
],
|
||||
id = actor,
|
||||
preferredUsername = handle,
|
||||
inbox = inbox or f"{actor}/inbox",
|
||||
outbox = outbox or f"{actor}/outbox",
|
||||
attachment = {},
|
||||
sharedInbox = {"endpoints": {"sharedInbox": shared_inbox or f"{proto}://{domain}/inbox"}},
|
||||
publicKey = {
|
||||
"id": f"{actor}#main-key",
|
||||
"owner": actor,
|
||||
"publicKeyPem": pubkey
|
||||
},
|
||||
**kwargs
|
||||
)
|
||||
|
||||
if fields:
|
||||
for key, value in fields.items():
|
||||
data.add_field(key, value)
|
||||
|
||||
if avatar:
|
||||
data.add_avatar(avatar)
|
||||
|
||||
if header:
|
||||
data.add_header(avatar)
|
||||
|
||||
return data
|
||||
|
||||
|
||||
@property
|
||||
def username(self) -> str:
|
||||
"Get the actor's username"
|
||||
return self["preferredUsername"]
|
||||
|
||||
|
||||
@property
|
||||
def keyid(self) -> str:
|
||||
"Get the ID of the actor's public key"
|
||||
return self["publicKey"]["id"]
|
||||
|
||||
|
||||
@property
|
||||
def pubkey(self) -> str:
|
||||
"Get the actor's PEM encoded public key"
|
||||
return self["publicKey"]["publicKeyPem"]
|
||||
|
||||
|
||||
@property
|
||||
def shared_inbox(self) -> str:
|
||||
"Get the instance's shared inbox"
|
||||
return self["endpoints"]["sharedInbox"]
|
||||
|
||||
|
||||
@property
|
||||
def signer(self) -> Signer:
|
||||
"Create :class:`Signer` object for the actor"
|
||||
if self._signer is None:
|
||||
self._signer = Signer.new_from_actor(self)
|
||||
|
||||
return self._signer
|
||||
|
||||
|
||||
@property
|
||||
def handle(self) -> str:
|
||||
"Get the full username that includes the domain"
|
||||
return f"{self.handle}@{self.domain}"
|
||||
|
||||
|
||||
def add_field(self, key: str, value: str) -> None:
|
||||
"""
|
||||
Add a profile field
|
||||
|
||||
:param key: Name of the field
|
||||
:param value: Value to associate with the name
|
||||
"""
|
||||
if "attachment" not in self:
|
||||
self.attachment = []
|
||||
|
||||
self.attachment.append(Attachment.new_field(key, value))
|
||||
|
||||
|
||||
def del_field(self, key: str) -> None:
|
||||
"""
|
||||
Delete a profile field
|
||||
|
||||
:param key: Name of the field
|
||||
"""
|
||||
self.attachment.remove(self.get_field(key))
|
||||
|
||||
|
||||
def get_field(self, key: str) -> dict[str, str]:
|
||||
"""
|
||||
Get a field dict with the specified name
|
||||
|
||||
:param key: Name of the field
|
||||
"""
|
||||
for field in self.get("attachment", []):
|
||||
if field["type"] == "PropertyValue" and field["name"] == key:
|
||||
return field
|
||||
|
||||
raise KeyError(key)
|
||||
|
||||
|
||||
@register_object_type(*ObjectType.Collection)
|
||||
class Object(Object):
|
||||
pass
|
||||
|
||||
|
||||
@register_object_type(*ObjectType.Media)
|
||||
class Media(Object):
|
||||
pass
|
||||
|
||||
|
||||
class Attachment(dict):
|
||||
@classmethod
|
||||
def new_field(cls: Type[Attachment], key: str, value: str) -> Attachment:
|
||||
"""
|
||||
Create a new attachment meant for storing a key/value pair on an actor
|
||||
|
||||
:param key: Name of the field
|
||||
:param value: Value to store with the field
|
||||
"""
|
||||
return cls({
|
||||
"type": "PropertyValue",
|
||||
"name": key,
|
||||
"value": value
|
||||
})
|
|
@ -6,9 +6,6 @@ build-backend = 'setuptools.build_meta'
|
|||
[tool.pylint.main]
|
||||
jobs = 0
|
||||
persistent = false
|
||||
ignore = [
|
||||
"message.py"
|
||||
]
|
||||
|
||||
[tool.pylint.design]
|
||||
max-args = 10
|
||||
|
|
Loading…
Reference in a new issue