forks/microblog.pub
1
0
Fork 1
mirror of https://git.sr.ht/~tsileo/microblog.pub synced 2024-11-15 03:04:28 +00:00
Code Issues Projects Releases Packages Wiki Activity

Tweak the activitypub helper

Browse source
This commit is contained in:
Thomas Sileo 2018-05-20 22:02:48 +02:00
parent eb9d7c6896
commit 2febca5711
2 changed files with 25 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
activitypub.py
View file

@ -162,7 +162,13 @@ class BaseActivity(object):
if len(set(kwargs.keys()) - set(allowed_keys)) > 0: if len(set(kwargs.keys()) - set(allowed_keys)) > 0:
raise ValueError('extra data left: {}'.format(kwargs)) raise ValueError('extra data left: {}'.format(kwargs))
else: else:
self._data.update(**kwargs) # Remove keys with `None` value
valid_kwargs = {}
for k, v in kwargs.items():
if v is None:
break
valid_kwargs[k] = v
self._data.update(**valid_kwargs)
def _init(self, **kwargs) -> Optional[List[str]]: def _init(self, **kwargs) -> Optional[List[str]]:
raise NotImplementedError raise NotImplementedError
@ -190,8 +196,15 @@ class BaseActivity(object):
def type_enum(self) -> ActivityTypes: def type_enum(self) -> ActivityTypes:
return ActivityTypes(self.type) return ActivityTypes(self.type)
def _set_id(self, uri: str, obj_id: str) -> None:
raise NotImplementedError
def set_id(self, uri: str, obj_id: str) -> None: def set_id(self, uri: str, obj_id: str) -> None:
self._data['id'] = uri self._data['id'] = uri
try:
self._set_id(uri, obj_id)
except NotImplementedError:
pass
def _actor_id(self, obj: ObjectOrIDType) -> str: def _actor_id(self, obj: ObjectOrIDType) -> str:
if isinstance(obj, dict) and obj['type'] == ActivityTypes.PERSON.value: if isinstance(obj, dict) and obj['type'] == ActivityTypes.PERSON.value:
@ -612,11 +625,11 @@ class Update(BaseActivity):
# If the object is a Person, it means the profile was updated, we just refresh our local cache # If the object is a Person, it means the profile was updated, we just refresh our local cache
ACTOR_SERVICE.get(obj.id, reload_cache=True) ACTOR_SERVICE.get(obj.id, reload_cache=True)
def _post_to_outbox(self, obj_id, activity, recipients): def _post_to_outbox(self, obj_id: str, activity: ObjectType, recipients: List[str]) -> None:
obj = self.get_object() obj = self.get_object()
update_prefix = 'activity.object.' update_prefix = 'activity.object.'
update = {'$set': dict(), '$unset': dict()} update = {'$set': dict(), '$unset': dict()} # type: Dict[str, Any]
update['$set'][f'{update_prefix}updated'] = datetime.utcnow().replace(microsecond=0).isoformat() + 'Z' update['$set'][f'{update_prefix}updated'] = datetime.utcnow().replace(microsecond=0).isoformat() + 'Z'
for k, v in obj._data.items(): for k, v in obj._data.items():
if k in ['id', 'type']: if k in ['id', 'type']:
@ -638,7 +651,7 @@ class Create(BaseActivity):
ACTIVITY_TYPE = ActivityTypes.CREATE ACTIVITY_TYPE = ActivityTypes.CREATE
ALLOWED_OBJECT_TYPES = [ActivityTypes.NOTE] ALLOWED_OBJECT_TYPES = [ActivityTypes.NOTE]
def _set_id(self, uri, obj_id): def _set_id(self, uri: str, obj_id: str) -> None:
self._data['object']['id'] = uri + '/activity' self._data['object']['id'] = uri + '/activity'
self._data['object']['url'] = ID + '/' + self.get_object().type.lower() + '/' + obj_id self._data['object']['url'] = ID + '/' + self.get_object().type.lower() + '/' + obj_id

25
app.py
View file

@ -503,23 +503,14 @@ def new():
if tag['type'] == 'Mention': if tag['type'] == 'Mention':
cc.append(tag['href']) cc.append(tag['href'])
if reply: note = activitypub.Note(
note = activitypub.Note( cc=cc,
cc=cc, to=[to if to else config.AS_PUBLIC],
to=[to if to else config.AS_PUBLIC], content=content, # TODO(tsileo): handle markdown
content=content, # TODO(tsileo): handle markdown tag=tags,
tag=tags, source={'mediaType': 'text/markdown', 'content': source},
source={'mediaType': 'text/markdown', 'content': source}, inReplyTo=reply.id if reply else None
inReplyTo=reply.id, # FIXME(tsieo): support None for inReplyTo? )
)
else:
note = activitypub.Note(
cc=cc,
to=[to if to else config.AS_PUBLIC],
content=content, # TODO(tsileo): handle markdown
tag=tags,
source={'mediaType': 'text/markdown', 'content': source},
)
create = note.build_create() create = note.build_create()
print(create.to_dict()) print(create.to_dict())