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

Merge pull request #56 from tsileo/post-visibility

Browse source 
Post visibility
This commit is contained in:
Thomas Sileo 2019-07-16 21:08:05 +02:00 committed by GitHub
commit 974acabd19
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 537 additions and 126 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

204
activitypub.py
View file

@ -8,6 +8,7 @@ from typing import Any
from typing import Dict from typing import Dict
from typing import List from typing import List
from typing import Optional from typing import Optional
from urllib.parse import urlparse
from bson.objectid import ObjectId from bson.objectid import ObjectId
from cachetools import LRUCache from cachetools import LRUCache
@ -34,6 +35,7 @@ logger = logging.getLogger(__name__)
ACTORS_CACHE = LRUCache(maxsize=256) ACTORS_CACHE = LRUCache(maxsize=256)
MY_PERSON = ap.Person(**ME)
def _actor_to_meta(actor: ap.BaseActivity, with_inbox=False) -> Dict[str, Any]: def _actor_to_meta(actor: ap.BaseActivity, with_inbox=False) -> Dict[str, Any]:
@ -114,9 +116,20 @@ class MicroblogPubBackend(Backend):
def save(self, box: Box, activity: ap.BaseActivity) -> None: def save(self, box: Box, activity: ap.BaseActivity) -> None:
"""Custom helper for saving an activity to the DB.""" """Custom helper for saving an activity to the DB."""
is_public = True visibility = ap.get_visibility(activity)
if activity.has_type(ap.ActivityType.CREATE) and not activity.is_public(): is_public = False
is_public = False if visibility in [ap.Visibility.PUBLIC, ap.Visibility.UNLISTED]:
is_public = True
object_id = None
try:
object_id = activity.get_object_id()
except ValueError:
pass
object_visibility = None
if activity.has_type([ap.ActivityType.CREATE, ap.ActivityType.ANNOUNCE]):
object_visibility = ap.get_visibility(activity.get_object()).name
actor_id = activity.get_actor().id
DB.activities.insert_one( DB.activities.insert_one(
{ {
@ -124,7 +137,17 @@ class MicroblogPubBackend(Backend):
"activity": activity.to_dict(), "activity": activity.to_dict(),
"type": _to_list(activity.type), "type": _to_list(activity.type),
"remote_id": activity.id, "remote_id": activity.id,
"meta": {"undo": False, "deleted": False, "public": is_public}, "meta": {
"undo": False,
"deleted": False,
"public": is_public,
"server": urlparse(activity.id).netloc,
"visibility": visibility.name,
"actor_id": actor_id,
"object_id": object_id,
"object_visibility": object_visibility,
"poll_answer": False,
},
} }
) )
@ -183,10 +206,40 @@ class MicroblogPubBackend(Backend):
) )
) )
def _fetch_iri(self, iri: str) -> ap.ObjectType: def _fetch_iri(self, iri: str) -> ap.ObjectType: # noqa: C901
# Shortcut if the instance actor is fetched
if iri == ME["id"]: if iri == ME["id"]:
return ME return ME
# Internal collecitons handling
# Followers
if iri == MY_PERSON.followers:
followers = []
for data in DB.activities.find(
{
"box": Box.INBOX.value,
"type": ap.ActivityType.FOLLOW.value,
"meta.undo": False,
}
):
followers.append(data["meta"]["actor_id"])
return {"type": "Collection", "items": followers}
# Following
if iri == MY_PERSON.following:
following = []
for data in DB.activities.find(
{
"box": Box.OUTBOX.value,
"type": ap.ActivityType.FOLLOW.value,
"meta.undo": False,
}
):
following.append(data["meta"]["object_id"])
return {"type": "Collection", "items": following}
# TODO(tsileo): handle the liked collection too
# Check if the activity is owned by this server # Check if the activity is owned by this server
if iri.startswith(BASE_URL): if iri.startswith(BASE_URL):
is_a_note = False is_a_note = False
@ -207,40 +260,48 @@ class MicroblogPubBackend(Backend):
if data["meta"]["deleted"]: if data["meta"]["deleted"]:
raise ActivityGoneError(f"{iri} is gone") raise ActivityGoneError(f"{iri} is gone")
return data["activity"] return data["activity"]
obj = DB.activities.find_one({"meta.object_id": iri, "type": "Create"})
if obj:
if obj["meta"]["deleted"]:
raise ActivityGoneError(f"{iri} is gone")
return obj["meta"].get("object") or obj["activity"]["object"]
# Check if it's cached because it's a follower
# Remove extra info (like the key hash if any)
cleaned_iri = iri.split("#")[0]
actor = DB.activities.find_one(
{
"meta.actor_id": cleaned_iri,
"type": ap.ActivityType.FOLLOW.value,
"meta.undo": False,
}
)
if actor and actor["meta"].get("actor"):
return actor["meta"]["actor"]
# Check if it's cached because it's a following
actor2 = DB.activities.find_one(
{
"meta.object_id": cleaned_iri,
"type": ap.ActivityType.FOLLOW.value,
"meta.undo": False,
}
)
if actor2 and actor2["meta"].get("object"):
return actor2["meta"]["object"]
# Fetch the URL via HTTP # Fetch the URL via HTTP
logger.info(f"dereference {iri} via HTTP") logger.info(f"dereference {iri} via HTTP")
return super().fetch_iri(iri) return super().fetch_iri(iri)
def fetch_iri(self, iri: str, no_cache=False) -> ap.ObjectType: def fetch_iri(self, iri: str, no_cache=False) -> ap.ObjectType:
if iri == ME["id"]:
return ME
if iri in ACTORS_CACHE:
logger.info(f"{iri} found in cache")
return ACTORS_CACHE[iri]
# data = DB.actors.find_one({"remote_id": iri})
# if data:
# if ap._has_type(data["type"], ap.ACTOR_TYPES):
# logger.info(f"{iri} found in DB cache")
# ACTORS_CACHE[iri] = data["data"]
# return data["data"]
if not no_cache: if not no_cache:
# Fetch the activity by checking the local DB first
data = self._fetch_iri(iri) data = self._fetch_iri(iri)
else: else:
return super().fetch_iri(iri) data = super().fetch_iri(iri)
logger.debug(f"_fetch_iri({iri!r}) == {data!r}") logger.debug(f"_fetch_iri({iri!r}) == {data!r}")
if ap._has_type(data["type"], ap.ACTOR_TYPES):
logger.debug(f"caching actor {iri}")
# Cache the actor
DB.actors.update_one(
{"remote_id": iri},
{"$set": {"remote_id": iri, "data": data}},
upsert=True,
)
ACTORS_CACHE[iri] = data
return data return data
@ -373,37 +434,36 @@ class MicroblogPubBackend(Backend):
@ensure_it_is_me @ensure_it_is_me
def inbox_delete(self, as_actor: ap.Person, delete: ap.Delete) -> None: def inbox_delete(self, as_actor: ap.Person, delete: ap.Delete) -> None:
obj = delete.get_object() obj_id = delete.get_object_id()
logger.debug("delete object={obj!r}") logger.debug("delete object={obj_id}")
try:
obj = ap.fetch_remote_activity(obj_id)
logger.info(f"inbox_delete handle_replies obj={obj!r}")
in_reply_to = obj.get_in_reply_to() if obj.inReplyTo else None
if obj.has_type(ap.CREATE_TYPES):
in_reply_to = ap._get_id(
DB.activities.find_one(
{"meta.object_id": obj_id, "type": ap.ActivityType.CREATE.value}
)["activity"]["object"].get("inReplyTo")
)
if in_reply_to:
self._handle_replies_delete(as_actor, in_reply_to)
except Exception:
logger.exception(f"failed to handle delete replies for {obj_id}")
DB.activities.update_one( DB.activities.update_one(
{"activity.object.id": obj.id}, {"$set": {"meta.deleted": True}} {"meta.object_id": obj_id, "type": "Create"},
{"$set": {"meta.deleted": True}},
) )
logger.info(f"inbox_delete handle_replies obj={obj!r}") # Foce undo other related activities
in_reply_to = obj.get_in_reply_to() if obj.inReplyTo else None DB.activities.update({"meta.object_id": obj_id}, {"$set": {"meta.undo": True}})
if delete.get_object().ACTIVITY_TYPE != ap.ActivityType.NOTE:
in_reply_to = ap._get_id(
DB.activities.find_one(
{
"activity.object.id": delete.get_object().id,
"type": ap.ActivityType.CREATE.value,
}
)["activity"]["object"].get("inReplyTo")
)
# Fake a Undo so any related Like/Announce doesn't appear on the web UI
DB.activities.update(
{"meta.object.id": obj.id},
{"$set": {"meta.undo": True, "meta.extra": "object deleted"}},
)
if in_reply_to:
self._handle_replies_delete(as_actor, in_reply_to)
@ensure_it_is_me @ensure_it_is_me
def outbox_delete(self, as_actor: ap.Person, delete: ap.Delete) -> None: def outbox_delete(self, as_actor: ap.Person, delete: ap.Delete) -> None:
DB.activities.update_one( DB.activities.update(
{"activity.object.id": delete.get_object().id}, {"meta.object_id": delete.get_object_id()},
{"$set": {"meta.deleted": True}}, {"$set": {"meta.deleted": True, "meta.undo": True}},
) )
obj = delete.get_object() obj = delete.get_object()
if delete.get_object().ACTIVITY_TYPE != ap.ActivityType.NOTE: if delete.get_object().ACTIVITY_TYPE != ap.ActivityType.NOTE:
@ -416,11 +476,6 @@ class MicroblogPubBackend(Backend):
)["activity"] )["activity"]
).get_object() ).get_object()
DB.activities.update(
{"meta.object.id": obj.id},
{"$set": {"meta.undo": True, "meta.exta": "object deleted"}},
)
self._handle_replies_delete(as_actor, obj.get_in_reply_to()) self._handle_replies_delete(as_actor, obj.get_in_reply_to())
@ensure_it_is_me @ensure_it_is_me
@ -481,6 +536,15 @@ class MicroblogPubBackend(Backend):
@ensure_it_is_me @ensure_it_is_me
def outbox_create(self, as_actor: ap.Person, create: ap.Create) -> None: def outbox_create(self, as_actor: ap.Person, create: ap.Create) -> None:
obj = create.get_object()
# Flag the activity as a poll answer if needed
print(f"POLL ANSWER ChECK {obj.get_in_reply_to()} {obj.name} {obj.content}")
if obj.get_in_reply_to() and obj.name and not obj.content:
DB.activities.update_one(
{"remote_id": create.id}, {"$set": {"meta.poll_answer": True}}
)
self._handle_replies(as_actor, create) self._handle_replies(as_actor, create)
@ensure_it_is_me @ensure_it_is_me
@ -540,7 +604,13 @@ class MicroblogPubBackend(Backend):
DB.activities.update_one( DB.activities.update_one(
{"remote_id": create.id}, {"remote_id": create.id},
{"$set": {"meta.answer_to": question.id, "meta.stream": False}}, {
"$set": {
"meta.answer_to": question.id,
"meta.stream": False,
"meta.poll_answer": True,
}
},
) )
return None return None
@ -637,7 +707,13 @@ def gen_feed():
fg.logo(ME.get("icon", {}).get("url")) fg.logo(ME.get("icon", {}).get("url"))
fg.language("en") fg.language("en")
for item in DB.activities.find( for item in DB.activities.find(
{"box": Box.OUTBOX.value, "type": "Create", "meta.deleted": False}, limit=10 {
"box": Box.OUTBOX.value,
"type": "Create",
"meta.deleted": False,
"meta.public": True,
},
limit=10,
).sort("_id", -1): ).sort("_id", -1):
fe = fg.add_entry() fe = fg.add_entry()
fe.id(item["activity"]["object"].get("url")) fe.id(item["activity"]["object"].get("url"))
@ -651,7 +727,13 @@ def json_feed(path: str) -> Dict[str, Any]:
"""JSON Feed (https://jsonfeed.org/) document.""" """JSON Feed (https://jsonfeed.org/) document."""
data = [] data = []
for item in DB.activities.find( for item in DB.activities.find(
{"box": Box.OUTBOX.value, "type": "Create", "meta.deleted": False}, limit=10 {
"box": Box.OUTBOX.value,
"type": "Create",
"meta.deleted": False,
"meta.public": True,
},
limit=10,
).sort("_id", -1): ).sort("_id", -1):
data.append( data.append(
{ {

191
app.py
View file

@ -86,8 +86,8 @@ from config import VERSION_DATE
from config import _drop_db from config import _drop_db
from poussetaches import PousseTaches from poussetaches import PousseTaches
from tasks import Tasks from tasks import Tasks
from utils import parse_datetime
from utils import opengraph from utils import opengraph
from utils import parse_datetime
from utils.key import get_secret_key from utils.key import get_secret_key
from utils.lookup import lookup from utils.lookup import lookup
from utils.media import Kind from utils.media import Kind
@ -143,15 +143,14 @@ def inject_config():
notes_count = DB.activities.find( notes_count = DB.activities.find(
{"box": Box.OUTBOX.value, "$or": [q, {"type": "Announce", "meta.undo": False}]} {"box": Box.OUTBOX.value, "$or": [q, {"type": "Announce", "meta.undo": False}]}
).count() ).count()
with_replies_count = DB.activities.find( # FIXME(tsileo): rename to all_count, and remove poll answers from it
{ all_q = {
"box": Box.OUTBOX.value, "box": Box.OUTBOX.value,
"type": {"$in": [ActivityType.CREATE.value, ActivityType.ANNOUNCE.value]}, "type": {"$in": [ActivityType.CREATE.value, ActivityType.ANNOUNCE.value]},
"meta.undo": False, "meta.undo": False,
"meta.deleted": False, "meta.deleted": False,
"meta.public": True, "meta.poll_answer": False,
} }
).count()
liked_count = DB.activities.count( liked_count = DB.activities.count(
{ {
"box": Box.OUTBOX.value, "box": Box.OUTBOX.value,
@ -181,7 +180,7 @@ def inject_config():
following_count=DB.activities.count(following_q) if logged_in else 0, following_count=DB.activities.count(following_q) if logged_in else 0,
notes_count=notes_count, notes_count=notes_count,
liked_count=liked_count, liked_count=liked_count,
with_replies_count=with_replies_count if logged_in else 0, with_replies_count=DB.activities.count(all_q) if logged_in else 0,
me=ME, me=ME,
base_url=config.BASE_URL, base_url=config.BASE_URL,
) )
@ -248,6 +247,19 @@ def _get_file_url(url, size, kind):
return url return url
@app.template_filter()
def visibility(v: str) -> str:
try:
return ap.Visibility[v].value.lower()
except Exception:
return v
@app.template_filter()
def visibility_is_public(v: str) -> bool:
return v in [ap.Visibility.PUBLIC.name, ap.Visibility.UNLISTED.name]
@app.template_filter() @app.template_filter()
def emojify(text): def emojify(text):
return emoji_unicode.replace( return emoji_unicode.replace(
@ -762,7 +774,13 @@ def authorize_follow():
if DB.activities.count(q) > 0: if DB.activities.count(q) > 0:
return redirect("/following") return redirect("/following")
follow = ap.Follow(actor=MY_PERSON.id, object=actor) follow = ap.Follow(
actor=MY_PERSON.id,
object=actor,
to=[actor],
cc=[ap.AS_PUBLIC],
published=ap.format_datetime(datetime.now(timezone.utc)),
)
post_to_outbox(follow) post_to_outbox(follow)
return redirect("/following") return redirect("/following")
@ -875,6 +893,7 @@ def index():
"activity.object.inReplyTo": None, "activity.object.inReplyTo": None,
"meta.deleted": False, "meta.deleted": False,
"meta.undo": False, "meta.undo": False,
"meta.public": True,
"$or": [{"meta.pinned": False}, {"meta.pinned": {"$exists": False}}], "$or": [{"meta.pinned": False}, {"meta.pinned": {"$exists": False}}],
} }
print(list(DB.activities.find(q))) print(list(DB.activities.find(q)))
@ -887,6 +906,7 @@ def index():
"type": ActivityType.CREATE.value, "type": ActivityType.CREATE.value,
"meta.deleted": False, "meta.deleted": False,
"meta.undo": False, "meta.undo": False,
"meta.public": True,
"meta.pinned": True, "meta.pinned": True,
} }
pinned = list(DB.activities.find(q_pinned)) pinned = list(DB.activities.find(q_pinned))
@ -906,15 +926,15 @@ def index():
return resp return resp
@app.route("/with_replies") @app.route("/all")
@login_required @login_required
def with_replies(): def all():
q = { q = {
"box": Box.OUTBOX.value, "box": Box.OUTBOX.value,
"type": {"$in": [ActivityType.CREATE.value, ActivityType.ANNOUNCE.value]}, "type": {"$in": [ActivityType.CREATE.value, ActivityType.ANNOUNCE.value]},
"meta.deleted": False, "meta.deleted": False,
"meta.public": True,
"meta.undo": False, "meta.undo": False,
"meta.poll_answer": False,
} }
outbox_data, older_than, newer_than = paginated_query(DB.activities, q) outbox_data, older_than, newer_than = paginated_query(DB.activities, q)
@ -1217,7 +1237,7 @@ def outbox():
if request.method == "GET": if request.method == "GET":
if not is_api_request(): if not is_api_request():
abort(404) abort(404)
# TODO(tsileo): returns the whole outbox if authenticated # TODO(tsileo): returns the whole outbox if authenticated and look at OCAP support
q = { q = {
"box": Box.OUTBOX.value, "box": Box.OUTBOX.value,
"meta.deleted": False, "meta.deleted": False,
@ -1252,7 +1272,11 @@ def outbox():
@app.route("/outbox/<item_id>") @app.route("/outbox/<item_id>")
def outbox_detail(item_id): def outbox_detail(item_id):
doc = DB.activities.find_one( doc = DB.activities.find_one(
{"box": Box.OUTBOX.value, "remote_id": back.activity_url(item_id)} {
"box": Box.OUTBOX.value,
"remote_id": back.activity_url(item_id),
"meta.public": True,
}
) )
if not doc: if not doc:
abort(404) abort(404)
@ -1268,7 +1292,11 @@ def outbox_detail(item_id):
@app.route("/outbox/<item_id>/activity") @app.route("/outbox/<item_id>/activity")
def outbox_activity(item_id): def outbox_activity(item_id):
data = DB.activities.find_one( data = DB.activities.find_one(
{"box": Box.OUTBOX.value, "remote_id": back.activity_url(item_id)} {
"box": Box.OUTBOX.value,
"remote_id": back.activity_url(item_id),
"meta.public": True,
}
) )
if not data: if not data:
abort(404) abort(404)
@ -1294,6 +1322,7 @@ def outbox_activity_replies(item_id):
"box": Box.OUTBOX.value, "box": Box.OUTBOX.value,
"remote_id": back.activity_url(item_id), "remote_id": back.activity_url(item_id),
"meta.deleted": False, "meta.deleted": False,
"meta.public": True,
} }
) )
if not data: if not data:
@ -1304,6 +1333,7 @@ def outbox_activity_replies(item_id):
q = { q = {
"meta.deleted": False, "meta.deleted": False,
"meta.public": True,
"type": ActivityType.CREATE.value, "type": ActivityType.CREATE.value,
"activity.object.inReplyTo": obj.get_object().id, "activity.object.inReplyTo": obj.get_object().id,
} }
@ -1329,6 +1359,7 @@ def outbox_activity_likes(item_id):
"box": Box.OUTBOX.value, "box": Box.OUTBOX.value,
"remote_id": back.activity_url(item_id), "remote_id": back.activity_url(item_id),
"meta.deleted": False, "meta.deleted": False,
"meta.public": True,
} }
) )
if not data: if not data:
@ -1532,6 +1563,7 @@ def admin_new():
reply=reply_id, reply=reply_id,
content=content, content=content,
thread=thread, thread=thread,
visibility=ap.Visibility,
emojis=EMOJIS.split(" "), emojis=EMOJIS.split(" "),
) )
@ -1666,7 +1698,14 @@ def api_delete():
"""API endpoint to delete a Note activity.""" """API endpoint to delete a Note activity."""
note = _user_api_get_note(from_outbox=True) note = _user_api_get_note(from_outbox=True)
delete = ap.Delete(actor=ID, object=ap.Tombstone(id=note.id).to_dict(embed=True)) # Create the delete, same audience as the Create object
delete = ap.Delete(
actor=ID,
object=ap.Tombstone(id=note.id).to_dict(embed=True),
to=note.to,
cc=note.cc,
published=ap.format_datetime(datetime.now(timezone.utc)),
)
delete_id = post_to_outbox(delete) delete_id = post_to_outbox(delete)
@ -1678,7 +1717,17 @@ def api_delete():
def api_boost(): def api_boost():
note = _user_api_get_note() note = _user_api_get_note()
announce = note.build_announce(MY_PERSON) # Ensures the note visibility allow us to build an Announce (in respect to the post visibility)
if ap.get_visibility(note) not in [ap.Visibility.PUBLIC, ap.Visibility.UNLISTED]:
abort(400)
announce = ap.Announce(
actor=MY_PERSON.id,
object=note.id,
to=[MY_PERSON.followers, note.attributedTo],
cc=[ap.AS_PUBLIC],
published=ap.format_datetime(datetime.now(timezone.utc)),
)
announce_id = post_to_outbox(announce) announce_id = post_to_outbox(announce)
return _user_api_response(activity=announce_id) return _user_api_response(activity=announce_id)
@ -1714,7 +1763,28 @@ def api_vote():
def api_like(): def api_like():
note = _user_api_get_note() note = _user_api_get_note()
like = note.build_like(MY_PERSON) to = []
cc = []
note_visibility = ap.get_visibility(note)
if note_visibility == ap.Visibility.PUBLIC:
to = [ap.AS_PUBLIC]
cc = [ID + "/followers", note.get_actor().id]
elif note_visibility == ap.Visibility.UNLISTED:
to = [ID + "/followers", note.get_actor().id]
cc = [ap.AS_PUBLIC]
else:
to = [note.get_actor().id]
like = ap.Like(
object=note.id,
actor=MY_PERSON.id,
to=to,
cc=cc,
published=ap.format_datetime(datetime.now(timezone.utc)),
)
like_id = post_to_outbox(like) like_id = post_to_outbox(like)
return _user_api_response(activity=like_id) return _user_api_response(activity=like_id)
@ -1779,8 +1849,16 @@ def api_undo():
raise ActivityNotFoundError(f"cannot found {oid}") raise ActivityNotFoundError(f"cannot found {oid}")
obj = ap.parse_activity(doc.get("activity")) obj = ap.parse_activity(doc.get("activity"))
undo = ap.Undo(
actor=MY_PERSON.id,
object=obj.to_dict(embed=True, embed_object_id_only=True),
published=ap.format_datetime(datetime.now(timezone.utc)),
to=obj.to,
cc=obj.cc,
)
# FIXME(tsileo): detect already undo-ed and make this API call idempotent # FIXME(tsileo): detect already undo-ed and make this API call idempotent
undo = obj.build_undo()
undo_id = post_to_outbox(undo) undo_id = post_to_outbox(undo)
return _user_api_response(activity=undo_id) return _user_api_response(activity=undo_id)
@ -1828,6 +1906,7 @@ def admin_bookmarks():
@app.route("/inbox", methods=["GET", "POST"]) # noqa: C901 @app.route("/inbox", methods=["GET", "POST"]) # noqa: C901
def inbox(): def inbox():
# GET /inbox
if request.method == "GET": if request.method == "GET":
if not is_api_request(): if not is_api_request():
abort(404) abort(404)
@ -1846,6 +1925,7 @@ def inbox():
) )
) )
# POST/ inbox
try: try:
data = request.get_json(force=True) data = request.get_json(force=True)
except Exception: except Exception:
@ -1995,22 +2075,41 @@ def api_new_note():
except ValueError: except ValueError:
pass pass
visibility = ap.Visibility[
_user_api_arg("visibility", default=ap.Visibility.PUBLIC.name)
]
content, tags = parse_markdown(source) content, tags = parse_markdown(source)
to = request.args.get("to")
cc = [ID + "/followers"] to, cc = [], []
if visibility == ap.Visibility.PUBLIC:
to = [ap.AS_PUBLIC]
cc = [ID + "/followers"]
elif visibility == ap.Visibility.UNLISTED:
to = [ID + "/followers"]
cc = [ap.AS_PUBLIC]
elif visibility == ap.Visibility.FOLLOWERS_ONLY:
to = [ID + "/followers"]
cc = []
if _reply: if _reply:
reply = ap.fetch_remote_activity(_reply) reply = ap.fetch_remote_activity(_reply)
cc.append(reply.attributedTo) if visibility == ap.Visibility.DIRECT:
to.append(reply.attributedTo)
else:
cc.append(reply.attributedTo)
for tag in tags: for tag in tags:
if tag["type"] == "Mention": if tag["type"] == "Mention":
cc.append(tag["href"]) if visibility == ap.Visibility.DIRECT:
to.append(tag["href"])
else:
cc.append(tag["href"])
raw_note = dict( raw_note = dict(
attributedTo=MY_PERSON.id, attributedTo=MY_PERSON.id,
cc=list(set(cc)), cc=list(set(cc)),
to=[to if to else ap.AS_PUBLIC], to=list(set(to)),
content=content, content=content,
tag=tags, tag=tags,
source={"mediaType": "text/markdown", "content": source}, source={"mediaType": "text/markdown", "content": source},
@ -2143,7 +2242,13 @@ def api_follow():
if existing: if existing:
return _user_api_response(activity=existing["activity"]["id"]) return _user_api_response(activity=existing["activity"]["id"])
follow = ap.Follow(actor=MY_PERSON.id, object=actor) follow = ap.Follow(
actor=MY_PERSON.id,
object=actor,
to=[actor],
cc=[ap.AS_PUBLIC],
published=ap.format_datetime(datetime.now(timezone.utc)),
)
follow_id = post_to_outbox(follow) follow_id = post_to_outbox(follow)
return _user_api_response(activity=follow_id) return _user_api_response(activity=follow_id)
@ -2680,7 +2785,13 @@ def task_finish_post_to_inbox():
back.inbox_like(MY_PERSON, activity) back.inbox_like(MY_PERSON, activity)
elif activity.has_type(ap.ActivityType.FOLLOW): elif activity.has_type(ap.ActivityType.FOLLOW):
# Reply to a Follow with an Accept # Reply to a Follow with an Accept
accept = ap.Accept(actor=ID, object=activity.to_dict(embed=True)) accept = ap.Accept(
actor=ID,
object=activity.to_dict(),
to=[activity.get_actor().id],
cc=[ap.AS_PUBLIC],
published=ap.format_datetime(datetime.now(timezone.utc)),
)
post_to_outbox(accept) post_to_outbox(accept)
elif activity.has_type(ap.ActivityType.UNDO): elif activity.has_type(ap.ActivityType.UNDO):
obj = activity.get_object() obj = activity.get_object()
@ -2833,34 +2944,21 @@ def task_cache_actor() -> str:
actor = activity.get_actor() actor = activity.get_actor()
cache_actor_with_inbox = False
if activity.has_type(ap.ActivityType.FOLLOW): if activity.has_type(ap.ActivityType.FOLLOW):
if actor.id != ID: if actor.id == ID:
# It's a Follow from the Inbox
cache_actor_with_inbox = True
else:
# It's a new following, cache the "object" (which is the actor we follow) # It's a new following, cache the "object" (which is the actor we follow)
DB.activities.update_one( DB.activities.update_one(
{"remote_id": iri}, {"remote_id": iri},
{ {
"$set": { "$set": {
"meta.object": activitypub._actor_to_meta( "meta.object": activity.get_object().to_dict(embed=True)
activity.get_object()
)
} }
}, },
) )
# Cache the actor info # Cache the actor info
DB.activities.update_one( DB.activities.update_one(
{"remote_id": iri}, {"remote_id": iri}, {"$set": {"meta.actor": actor.to_dict(embed=True)}}
{
"$set": {
"meta.actor": activitypub._actor_to_meta(
actor, cache_actor_with_inbox
)
}
},
) )
app.logger.info(f"actor cached for {iri}") app.logger.info(f"actor cached for {iri}")
@ -2965,7 +3063,7 @@ def task_process_new_activity():
elif activity.has_type(ap.ActivityType.DELETE): elif activity.has_type(ap.ActivityType.DELETE):
note = DB.activities.find_one( note = DB.activities.find_one(
{"activity.object.id": activity.get_object().id} {"activity.object.id": activity.get_object_id()}
) )
if note and note["meta"].get("forwarded", False): if note and note["meta"].get("forwarded", False):
# If the activity was originally forwarded, forward the delete too # If the activity was originally forwarded, forward the delete too
@ -3093,6 +3191,7 @@ def task_fetch_remote_question():
} }
) )
remote_question = get_backend().fetch_iri(iri, no_cache=True) remote_question = get_backend().fetch_iri(iri, no_cache=True)
# FIXME(tsileo): compute and set `meta.object_visiblity` (also update utils.py to do it)
if ( if (
local_question local_question
and ( and (

152
migrations.py Normal file
View file

@ -0,0 +1,152 @@
"""Migrations that will be run automatically at startup."""
from typing import Any
from typing import Dict
from urllib.parse import urlparse
from little_boxes import activitypub as ap
from utils.migrations import DB
from utils.migrations import Migration
from utils.migrations import logger
from utils.migrations import perform # noqa: just here for export
from config import ID
import activitypub
back = activitypub.MicroblogPubBackend()
ap.use_backend(back)
class _1_MetaMigration(Migration):
"""Add new metadata to simplify querying."""
def __guess_visibility(self, data: Dict[str, Any]) -> ap.Visibility:
to = data.get("to", [])
cc = data.get("cc", [])
if ap.AS_PUBLIC in to:
return ap.Visibility.PUBLIC
elif ap.AS_PUBLIC in cc:
return ap.Visibility.UNLISTED
else:
# Uses a bit of heuristic here, it's too expensive to fetch the actor, so assume the followers
# collection has "/collection" in it (which is true for most software), and at worst, we will
# classify it as "DIRECT" which behave the same as "FOLLOWERS_ONLY" (i.e. no Announce)
followers_only = False
for item in to:
if "/followers" in item:
followers_only = True
break
if not followers_only:
for item in cc:
if "/followers" in item:
followers_only = True
break
if followers_only:
return ap.Visibility.FOLLOWERS_ONLY
return ap.Visibility.DIRECT
def migrate(self) -> None: # noqa: C901 # too complex
for data in DB.activities.find():
logger.info(f"before={data}")
obj = data["activity"].get("object")
set_meta: Dict[str, Any] = {}
# Set `meta.object_id` (str)
if not data["meta"].get("object_id"):
set_meta["meta.object_id"] = None
if obj:
if isinstance(obj, str):
set_meta["meta.object_id"] = data["activity"]["object"]
elif isinstance(obj, dict):
obj_id = obj.get("id")
if obj_id:
set_meta["meta.object_id"] = obj_id
# Set `meta.object_visibility` (str)
if not data["meta"].get("object_visibility"):
set_meta["meta.object_visibility"] = None
object_id = data["meta"].get("object_id") or set_meta.get(
"meta.object_id"
)
if object_id:
obj = data["meta"].get("object") or data["activity"].get("object")
if isinstance(obj, dict):
set_meta["meta.object_visibility"] = self.__guess_visibility(
obj
).name
# Set `meta.actor_id` (str)
if not data["meta"].get("actor_id"):
set_meta["meta.actor_id"] = None
actor = data["activity"].get("actor")
if actor:
if isinstance(actor, str):
set_meta["meta.actor_id"] = data["activity"]["actor"]
elif isinstance(actor, dict):
actor_id = actor.get("id")
if actor_id:
set_meta["meta.actor_id"] = actor_id
# Set `meta.poll_answer` (bool)
if not data["meta"].get("poll_answer"):
set_meta["meta.poll_answer"] = False
if obj:
if isinstance(obj, dict):
if (
obj.get("name")
and not obj.get("content")
and obj.get("inReplyTo")
):
set_meta["meta.poll_answer"] = True
# Set `meta.visibility` (str)
if not data["meta"].get("visibility"):
set_meta["meta.visibility"] = self.__guess_visibility(
data["activity"]
).name
if not data["meta"].get("server"):
set_meta["meta.server"] = urlparse(data["remote_id"]).netloc
logger.info(f"meta={set_meta}\n")
if set_meta:
DB.activities.update_one({"_id": data["_id"]}, {"$set": set_meta})
class _2_FollowMigration(Migration):
"""Add new metadata to update the cached actor in Follow activities."""
def migrate(self) -> None:
actor_cache: Dict[str, Dict[str, Any]] = {}
for data in DB.activities.find({"type": ap.ActivityType.FOLLOW.value}):
if data["meta"]["actor_id"] == ID:
# It's a "following"
actor = actor_cache.get(data["meta"]["object_id"])
if not actor:
actor = ap.parse_activity(
ap.get_backend().fetch_iri(
data["meta"]["object_id"], no_cache=True
)
).to_dict(embed=True)
if not actor:
raise ValueError(f"missing actor {data!r}")
actor_cache[actor["id"]] = actor
DB.activities.update_one(
{"_id": data["_id"]}, {"$set": {"meta.object": actor}}
)
else:
# It's a "followers"
actor = actor_cache.get(data["meta"]["actor_id"])
if not actor:
actor = ap.parse_activity(
ap.get_backend().fetch_iri(
data["meta"]["actor_id"], no_cache=True
)
).to_dict(embed=True)
if not actor:
raise ValueError(f"missing actor {data!r}")
actor_cache[actor["id"]] = actor
DB.activities.update_one(
{"_id": data["_id"]}, {"$set": {"meta.actor": actor}}
)

1
run.sh
View file

@ -1,3 +1,4 @@
#!/bin/bash #!/bin/bash
python -c "import logging; logging.basicConfig(level=logging.DEBUG); import migrations; migrations.perform()"
python -c "import config; config.create_indexes()" python -c "import config; config.create_indexes()"
gunicorn -t 600 -w 5 -b 0.0.0.0:5005 --log-level debug app:app gunicorn -t 600 -w 5 -b 0.0.0.0:5005 --log-level debug app:app

10
sass/base_theme.scss
View file

@ -274,6 +274,16 @@ a:hover {
background: $primary-color; background: $primary-color;
color: $background-color; color: $background-color;
} }
.bar-item-no-border {
color: $color-light;
background: inherit;
cursor: default;
}
.bar-item-no-border:hover {
color: $color-light;
background: inherit;
cursor: default;
}
button.bar-item { button.bar-item {
border: 0 border: 0
} }

2
tasks.py
View file

@ -2,8 +2,8 @@ import os
from datetime import datetime from datetime import datetime
from datetime import timezone from datetime import timezone
from utils import parse_datetime
from poussetaches import PousseTaches from poussetaches import PousseTaches
from utils import parse_datetime
p = PousseTaches( p = PousseTaches(
os.getenv("MICROBLOGPUB_POUSSETACHES_HOST", "http://localhost:7991"), os.getenv("MICROBLOGPUB_POUSSETACHES_HOST", "http://localhost:7991"),

2
templates/header.html
View file

@ -18,7 +18,7 @@
<div class="menu"> <div class="menu">
<ul> <ul>
<li><a href="/" {% if request.path == "/" %}class="selected"{% endif %}>Notes <small class="badge">{{ notes_count }}</small></a></li> <li><a href="/" {% if request.path == "/" %}class="selected"{% endif %}>Notes <small class="badge">{{ notes_count }}</small></a></li>
{% if session.logged_in %}<li><a href="/with_replies" {% if request.path == "/with_replies" %}class="selected"{% endif %}>With replies <small class="badge">{{ with_replies_count }}</small></a></li> {% if session.logged_in %}<li><a href="/all" {% if request.path == url_for("all") %}class="selected"{% endif %}>All <small class="badge">{{ with_replies_count }}</small></a></li>
{% endif %} {% endif %}
<li><a href="/liked" {% if request.path == "/liked" %}class="selected"{% endif %}>Liked <small class="badge">{{ liked_count }}</small></a></li> <li><a href="/liked" {% if request.path == "/liked" %}class="selected"{% endif %}>Liked <small class="badge">{{ liked_count }}</small></a></li>
<li><a href="/followers"{% if request.path == "/followers" %} class="selected" {% endif %}>Followers <small class="badge">{{ followers_count }}</small></a></li> <li><a href="/followers"{% if request.path == "/followers" %} class="selected" {% endif %}>Followers <small class="badge">{{ followers_count }}</small></a></li>

2
templates/index.html
View file

@ -53,7 +53,7 @@
{% endif %} {% endif %}
{% endif %} {% endif %}
{% if item.meta.object %} {% if item.meta.object %}
{{ utils.display_note(item.meta.object, ui=False, meta={'actor': item.meta.object_actor}) }} {{ utils.display_note(item.meta.object, ui=False, meta=item.meta) }}
{% endif %} {% endif %}
{% elif item | has_type('Create') %} {% elif item | has_type('Create') %}
{{ utils.display_note(item.activity.object, meta=item.meta, no_color=True) }} {{ utils.display_note(item.activity.object, meta=item.meta, no_color=True) }}

1
templates/layout.html
View file

@ -12,6 +12,7 @@
{% block links %}{% endblock %} {% block links %}{% endblock %}
{% if config.THEME_COLOR %}<meta name="theme-color" content="{{ config.THEME_COLOR }}">{% endif %} {% if config.THEME_COLOR %}<meta name="theme-color" content="{{ config.THEME_COLOR }}">{% endif %}
<style>{{ config.CSS | safe }} <style>{{ config.CSS | safe }}
.icon { color: #555; }
.emoji { .emoji {
width: 20px; width: 20px;
} }

2
templates/liked.html
View file

@ -20,7 +20,7 @@
{% endif %} {% endif %}
{% if item.meta.object %} {% if item.meta.object %}
{{ utils.display_note(item.meta.object, meta={'actor': item.meta.object_actor}) }} {{ utils.display_note(item.meta.object, meta=item.meta) }}
{% endif %} {% endif %}
{% endfor %} {% endfor %}

5
templates/new.html
View file

@ -21,6 +21,11 @@
<form action="/api/new_{% if request.args.get("question") == "1" %}question{%else%}note{%endif%}" method="POST" enctype="multipart/form-data"> <form action="/api/new_{% if request.args.get("question") == "1" %}question{%else%}note{%endif%}" method="POST" enctype="multipart/form-data">
<input type="hidden" name="redirect" value="/"> <input type="hidden" name="redirect" value="/">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}"> <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<select name="visibility">
{% for v in visibility %}
<option value="{{v.name}}">{{ v.value }}</option>
{% endfor %}
</select>
{% if reply %}<input type="hidden" name="reply" value="{{reply}}">{% endif %} {% if reply %}<input type="hidden" name="reply" value="{{reply}}">{% endif %}
<p> <p>

6
templates/stream.html
View file

@ -18,7 +18,7 @@
<p style="margin-left:70px;padding-bottom:5px;display:inline-block;"><span class="bar-item-no-hover"><a style="color:#808080;" href="{{ boost_actor.url | get_url }}">{{ boost_actor.name or boost_actor.preferredUsername }}</a> boosted</span></p> <p style="margin-left:70px;padding-bottom:5px;display:inline-block;"><span class="bar-item-no-hover"><a style="color:#808080;" href="{{ boost_actor.url | get_url }}">{{ boost_actor.name or boost_actor.preferredUsername }}</a> boosted</span></p>
{% endif %} {% endif %}
{% if item.meta.object %} {% if item.meta.object %}
{{ utils.display_note(item.meta.object, ui=True) }} {{ utils.display_note(item.meta.object, ui=True, meta=item.meta) }}
{% endif %} {% endif %}
{% endif %} {% endif %}
@ -26,7 +26,7 @@
{% set boost_actor = item.meta.actor %} {% set boost_actor = item.meta.actor %}
<p style="margin-left:70px;padding-bottom:5px;display:inline-block;"><span class="bar-item-no-hover"><a style="color:#808080;" href="{{ boost_actor.url | get_url }}">{{ boost_actor.name or boost_actor.preferredUsername }}</a> liked</span></p> <p style="margin-left:70px;padding-bottom:5px;display:inline-block;"><span class="bar-item-no-hover"><a style="color:#808080;" href="{{ boost_actor.url | get_url }}">{{ boost_actor.name or boost_actor.preferredUsername }}</a> liked</span></p>
{% if item.meta.object %} {% if item.meta.object %}
{{ utils.display_note(item.meta.object, ui=False, meta={'actor': item.meta.object_actor}) }} {{ utils.display_note(item.meta.object, ui=False, meta=item.meta) }}
{% endif %} {% endif %}
{% endif %} {% endif %}
@ -58,7 +58,7 @@
{% if item | has_type('question_ended') %} {% if item | has_type('question_ended') %}
<p style="margin-left:70px;padding-bottom:5px;display:inline-block;"><span class="bar-item-no-hover">poll ended</span></p> <p style="margin-left:70px;padding-bottom:5px;display:inline-block;"><span class="bar-item-no-hover">poll ended</span></p>
{{ utils.display_note(item.activity) }} {{ utils.display_note(item.activity, meta={"object_visibility": "PUBLIC"}) }}
{% endif %} {% endif %}
{% endif %} {% endif %}

29
templates/utils.html
View file

@ -17,7 +17,9 @@
{% macro display_note(obj, perma=False, ui=False, likes=[], shares=[], meta={}, no_color=False) -%} {% macro display_note(obj, perma=False, ui=False, likes=[], shares=[], meta={}, no_color=False) -%}
{% if meta.actor %} {% if meta.object_actor %}
{% set actor = meta.object_actor %}
{% elif meta.actor %}
{% set actor = meta.actor %} {% set actor = meta.actor %}
{% else %} {% else %}
{% set actor = obj.attributedTo | get_actor %} {% set actor = obj.attributedTo | get_actor %}
@ -51,11 +53,11 @@
<div class="note-wrapper"> <div class="note-wrapper">
<div style="clear:both;height:20px;"> <div style="clear:both;height:20px;">
<a href="{{ actor | url_or_id | get_url }}" style="margin:0;text-decoration:none;margin: 0;text-decoration: none;display: block;width: 80%;overflow: hidden;white-space: nowrap;text-overflow: ellipsis;float: left;" class="no-hover"><strong>{{ actor.name or actor.preferredUsername }}</strong> <a href="{{ actor | url_or_id | get_url }}" style="margin:0;text-decoration:none;margin: 0;text-decoration: none;display: block;width: 75%;overflow: hidden;white-space: nowrap;text-overflow: ellipsis;float: left;" class="no-hover"><strong>{{ actor.name or actor.preferredUsername }}</strong>
<span class="l">@{% if not no_color and obj.id | is_from_outbox %}<span class="pcolor">{{ actor.preferredUsername }}</span>{% else %}{{ actor.preferredUsername }}{% endif %}@{% if not no_color and obj.id | is_from_outbox %}<span class="pcolor">{{ actor | url_or_id | get_url | domain }}</span>{% else %}{{ actor | url_or_id | get_url | domain }}{% endif %}</span></a> <span class="l">@{% if not no_color and obj.id | is_from_outbox %}<span class="pcolor">{{ actor.preferredUsername }}</span>{% else %}{{ actor.preferredUsername }}{% endif %}@{% if not no_color and obj.id | is_from_outbox %}<span class="pcolor">{{ actor | url_or_id | get_url | domain }}</span>{% else %}{{ actor | url_or_id | get_url | domain }}{% endif %}</span></a>
{% if not perma %} {% if not perma %}
<span style="float:right;width: 20%;text-align: right;overflow: hidden;white-space: nowrap;text-overflow: ellipsis;display: block;"> <span style="float:right;width: 25%;text-align: right;overflow: hidden;white-space: nowrap;text-overflow: ellipsis;display: block;">
<a rel="noopener" class="u-url u-uid note-permalink l" href="{{ obj | url_or_id | get_url }}"> <a rel="noopener" class="u-url u-uid note-permalink l" href="{{ obj | url_or_id | get_url }}">
<time class="dt-published" title="{{ obj.published }}" datetime="{{ obj.published }}">{{ obj.published | format_timeago }}</time></a> <time class="dt-published" title="{{ obj.published }}" datetime="{{ obj.published }}">{{ obj.published | format_timeago }}</time></a>
</span> </span>
@ -163,23 +165,21 @@
<div class="bottom-bar"> <div class="bottom-bar">
{% if perma %} {% if perma %}
<span class="perma-item" style="float:left;padding:5px;">{{ obj.published | format_time }}</span> <span class="perma-item" style="float:left;padding:5px;">{{ obj.published | format_time }}</span>
{% if not (obj.id | is_from_outbox) %}
<a class ="bar-item" href="{{ obj | url_or_id | get_url }}">permalink</a>
{% endif %}
{% else %}
<a class ="bar-item" href="{{ obj | url_or_id | get_url }}">permalink</a>
{% endif %} {% endif %}
{% if meta.count_reply and obj.id | is_from_outbox %}<a class ="bar-item" href="{{ obj.url | get_url }}"><strong>{{ meta.count_reply }}</strong> replies</a> {% if meta.count_reply and obj.id | is_from_outbox %}<a class ="bar-item" href="{{ obj.url | get_url }}"><strong>{{ meta.count_reply }}</strong> replies</a>
{% elif meta.count_reply and session.logged_in %} {% elif meta.count_reply and session.logged_in %}
<a class ="bar-item" href="/admin/thread?oid={{aid}}"><strong>{{ meta.count_reply }}</strong> replies</a>{% endif %} <a class="bar-item" href="/admin/thread?oid={{aid}}"><strong>{{ meta.count_reply }}</strong> replies</a>{% endif %}
{% if not perma and meta.count_boost and obj.id | is_from_outbox %}<a class ="bar-item" href="{{ obj.url | get_url }}"><strong>{{ meta.count_boost }}</strong> boosts</a>{% endif %} {% if not perma and meta.count_boost and obj.id | is_from_outbox %}<a class ="bar-item" href="{{ obj.url | get_url }}"><strong>{{ meta.count_boost }}</strong> boosts</a>{% endif %}
{% if not perma and meta.count_like and obj.id | is_from_outbox %}<a class ="bar-item" href="{{ obj.url | get_url }}"><strong>{{ meta.count_like }}</strong> likes</a>{% endif %} {% if not perma and meta.count_like and obj.id | is_from_outbox %}<a class ="bar-item" href="{{ obj.url | get_url }}"><strong>{{ meta.count_like }}</strong> likes</a>{% endif %}
{% if session.logged_in %} {% if session.logged_in %}
{% if ui%} {% if ui%}
<a class="bar-item" href="/admin/new?reply={{ aid }}">reply</a>
{% if meta.object_visibility | visibility_is_public %}
{% if meta.boosted %} {% if meta.boosted %}
<form action="/api/undo" class="action-form" method="POST"> <form action="/api/undo" class="action-form" method="POST">
<input type="hidden" name="redirect" value="{{ redir }}"> <input type="hidden" name="redirect" value="{{ redir }}">
@ -195,6 +195,7 @@
<button type="submit" class="bar-item">boost</button> <button type="submit" class="bar-item">boost</button>
</form> </form>
{% endif %} {% endif %}
{% endif %}
{% if meta.liked %} {% if meta.liked %}
<form action="/api/undo" class="action-form" method="POST"> <form action="/api/undo" class="action-form" method="POST">
@ -237,7 +238,7 @@
<input type="hidden" name="redirect" value="{{ redir }}"> <input type="hidden" name="redirect" value="{{ redir }}">
<input type="hidden" name="id" value="{{ obj.id }}"> <input type="hidden" name="id" value="{{ obj.id }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}"> <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bar-item">delete</button> <button type="submit" class="bar-item" onclick="return confirm('Confirm the delete action?');">delete</button>
</form> </form>
{% if meta.pinned %} {% if meta.pinned %}
<form action="/api/note/unpin" class="action-form" method="POST"> <form action="/api/note/unpin" class="action-form" method="POST">
@ -260,12 +261,16 @@
<input type="hidden" name="redirect" value="{{ redir }}"> <input type="hidden" name="redirect" value="{{ redir }}">
<input type="hidden" name="actor" value="{{ actor.id }}"> <input type="hidden" name="actor" value="{{ actor.id }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}"> <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bar-item">block</button> <button type="submit" class="bar-item" onclick="return confirm('Confirm the block action?');">block</button>
</form> </form>
{% endif %} {% endif %}
<a class="bar-item" href="/admin/new?reply={{ aid }}">reply</a>
{% endif %} {% endif %}
<a class="bar-item" href="{{ obj | url_or_id | get_url }}">permalink</a>
{% if session.logged_in %}
<a class="bar-item bar-item-no-border">{{ meta.object_visibility | visibility }}</a>
{% endif %}
</div> </div>

56
utils/migrations.py Normal file
View file

@ -0,0 +1,56 @@
"""Automatic migration tools for the da:ta stored in MongoDB."""
import logging
from abc import ABC
from abc import abstractmethod
from typing import List
from typing import Type
from config import DB
logger = logging.getLogger(__name__)
# Used to keep track of all the defined migrations
_MIGRATIONS: List[Type["Migration"]] = []
def perform() -> None:
"""Perform all the defined migration."""
for migration in _MIGRATIONS:
migration().perform()
class Migration(ABC):
"""Abstract class for migrations."""
def __init__(self) -> None:
self.name = self.__class__.__qualname__
self._col = DB.migrations
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
_MIGRATIONS.append(cls)
def _apply(self) -> None:
self._col.insert_one({"name": self.name})
def _reset(self) -> None:
self._col.delete_one({"name": self.name})
def _is_applied(self) -> bool:
return bool(self._col.find_one({"name": self.name}))
@abstractmethod
def migrate(self) -> None:
"""Expected to be implemented by actual migrations."""
pass
def perform(self) -> None:
if self._is_applied():
logger.info(f"Skipping migration {self.name} (already applied)")
return
logger.info(f"Performing migration {self.name}...")
self.migrate()
self._apply()
logger.info("Done")