1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
from typing import Literal
from django.forms import ValidationError
from django.shortcuts import get_object_or_404
from ninja import Schema
from activities.models import (
Post,
PostAttachment,
PostInteraction,
PostStates,
TimelineEvent,
)
from api import schemas
from api.views.base import api_router
from core.models import Config
from ..decorators import identity_required
class PostStatusSchema(Schema):
status: str
in_reply_to_id: str | None = None
sensitive: bool = False
spoiler_text: str | None = None
visibility: Literal["public", "unlisted", "private", "direct"] = "public"
language: str | None = None
scheduled_at: str | None = None
media_ids: list[str] = []
@api_router.post("/v1/statuses", response=schemas.Status)
@identity_required
def post_status(request, details: PostStatusSchema):
# Check text length
if len(details.status) > Config.system.post_length:
raise ValidationError("Status is too long")
if len(details.status) == 0 and not details.media_ids:
raise ValidationError("Status is empty")
# Grab attachments
attachments = [get_object_or_404(PostAttachment, pk=id) for id in details.media_ids]
# Create the Post
visibility_map = {
"public": Post.Visibilities.public,
"unlisted": Post.Visibilities.unlisted,
"private": Post.Visibilities.followers,
"direct": Post.Visibilities.mentioned,
}
reply_post = None
if details.in_reply_to_id:
try:
reply_post = Post.objects.get(pk=details.in_reply_to_id)
except Post.DoesNotExist:
pass
post = Post.create_local(
author=request.identity,
content=details.status,
summary=details.spoiler_text,
sensitive=details.sensitive,
visibility=visibility_map[details.visibility],
reply_to=reply_post,
attachments=attachments,
)
# Add their own timeline event for immediate visibility
TimelineEvent.add_post(request.identity, post)
return post.to_mastodon_json()
@api_router.get("/v1/statuses/{id}", response=schemas.Status)
@identity_required
def status(request, id: str):
post = get_object_or_404(Post, pk=id)
interactions = PostInteraction.get_post_interactions([post], request.identity)
return post.to_mastodon_json(interactions=interactions)
@api_router.delete("/v1/statuses/{id}", response=schemas.Status)
@identity_required
def delete_status(request, id: str):
post = get_object_or_404(Post, pk=id)
post.transition_perform(PostStates.deleted)
TimelineEvent.objects.filter(subject_post=post, identity=request.identity).delete()
return post.to_mastodon_json()
@api_router.get("/v1/statuses/{id}/context", response=schemas.Context)
@identity_required
def status_context(request, id: str):
post = get_object_or_404(Post, pk=id)
parent = post.in_reply_to_post()
ancestors = []
if parent:
ancestors.append(parent)
descendants = list(Post.objects.filter(in_reply_to=post.object_uri)[:40])
interactions = PostInteraction.get_post_interactions(
[post] + ancestors + descendants, request.identity
)
return {
"ancestors": [p.to_mastodon_json(interactions=interactions) for p in ancestors],
"descendants": [
p.to_mastodon_json(interactions=interactions) for p in descendants
],
}
@api_router.post("/v1/statuses/{id}/favourite", response=schemas.Status)
@identity_required
def favourite_status(request, id: str):
post = get_object_or_404(Post, pk=id)
post.like_as(request.identity)
interactions = PostInteraction.get_post_interactions([post], request.identity)
return post.to_mastodon_json(interactions=interactions)
@api_router.post("/v1/statuses/{id}/unfavourite", response=schemas.Status)
@identity_required
def unfavourite_status(request, id: str):
post = get_object_or_404(Post, pk=id)
post.unlike_as(request.identity)
interactions = PostInteraction.get_post_interactions([post], request.identity)
return post.to_mastodon_json(interactions=interactions)
@api_router.post("/v1/statuses/{id}/reblog", response=schemas.Status)
@identity_required
def reblog_status(request, id: str):
post = get_object_or_404(Post, pk=id)
post.boost_as(request.identity)
interactions = PostInteraction.get_post_interactions([post], request.identity)
return post.to_mastodon_json(interactions=interactions)
@api_router.post("/v1/statuses/{id}/unreblog", response=schemas.Status)
@identity_required
def unreblog_status(request, id: str):
post = get_object_or_404(Post, pk=id)
post.unboost_as(request.identity)
interactions = PostInteraction.get_post_interactions([post], request.identity)
return post.to_mastodon_json(interactions=interactions)
|
