Python script, ASCII text executable
1
__all__ = [
2
"RepoAccess",
3
"RepoFavourite",
4
"Repo",
5
"UserFollow",
6
"UserNotification",
7
"User",
8
"Notification",
9
"PostVote",
10
"Post",
11
"Commit",
12
"PullRequest",
13
"EmailChangeRequest",
14
]
15
import secrets
17
import subprocess
18
from app import app, db, bcrypt
19
import git
20
from datetime import datetime, timedelta
21
from enum import Enum
22
from PIL import Image
23
from cairosvg import svg2png
24
import os
25
import config
26
import cairosvg
27
import random
28
import celery_tasks
29
with (app.app_context()):
31
class RepoAccess(db.Model):
32
id = db.Column(db.Integer, primary_key=True)
33
user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False)
34
repo_route = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False)
35
access_level = db.Column(db.SmallInteger(), nullable=False) # 0 read-only, 1 read-write, 2 admin
36
user = db.relationship("User", back_populates="repo_access")
38
repo = db.relationship("Repo", back_populates="repo_access")
39
__table_args__ = (db.UniqueConstraint("user_username", "repo_route", name="_user_repo_uc"),)
41
def __init__(self, user, repo, level):
43
self.user_username = user.username
44
self.repo_route = repo.route
45
self.access_level = level
46
class RepoFavourite(db.Model):
49
id = db.Column(db.Integer, primary_key=True)
50
user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False)
51
repo_route = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False)
52
notify_commit = db.Column(db.Boolean, default=False, nullable=False)
54
notify_forum = db.Column(db.Boolean, default=False, nullable=False)
55
notify_pr = db.Column(db.Boolean, default=False, nullable=False)
56
notify_admin = db.Column(db.Boolean, default=False, nullable=False)
57
user = db.relationship("User", back_populates="favourites")
59
repo = db.relationship("Repo", back_populates="favourites")
60
__table_args__ = (db.UniqueConstraint("user_username", "repo_route", name="_user_repo_uc1"),)
62
def __init__(self, user, repo):
64
self.user_username = user.username
65
self.repo_route = repo.route
66
class PostVote(db.Model):
69
id = db.Column(db.Integer, primary_key=True)
70
user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False)
71
post_identifier = db.Column(db.String(109), db.ForeignKey("post.identifier"), nullable=False)
72
vote_score = db.Column(db.SmallInteger(), nullable=False)
73
user = db.relationship("User", back_populates="votes")
75
post = db.relationship("Post", back_populates="votes")
76
__table_args__ = (db.UniqueConstraint("user_username", "post_identifier", name="_user_post_uc"),)
78
def __init__(self, user, post, score):
80
self.user_username = user.username
81
self.post_identifier = post.identifier
82
self.vote_score = score
83
class User(db.Model):
86
username = db.Column(db.String(32), unique=True, nullable=False, primary_key=True)
87
display_name = db.Column(db.Unicode(128), unique=False, nullable=True)
88
bio = db.Column(db.Unicode(16384), unique=False, nullable=True)
89
password_hashed = db.Column(db.String(60), nullable=False)
90
email = db.Column(db.String(254), nullable=True)
91
company = db.Column(db.Unicode(64), nullable=True)
92
company_url = db.Column(db.String(256), nullable=True)
93
url = db.Column(db.String(256), nullable=True)
94
show_mail = db.Column(db.Boolean, default=False, nullable=False)
95
location = db.Column(db.Unicode(64), nullable=True)
96
creation_date = db.Column(db.DateTime, default=datetime.utcnow)
97
default_page_length = db.Column(db.SmallInteger, nullable=False, default=32, server_default="32")
98
max_post_nesting = db.Column(db.SmallInteger, nullable=False, default=3, server_default="3")
99
repositories = db.relationship("Repo", back_populates="owner")
101
followers = db.relationship("UserFollow", back_populates="followed", foreign_keys="[UserFollow.followed_username]")
102
follows = db.relationship("UserFollow", back_populates="follower", foreign_keys="[UserFollow.follower_username]")
103
email_change_requests = db.relationship("EmailChangeRequest", back_populates="user")
104
repo_access = db.relationship("RepoAccess", back_populates="user")
105
votes = db.relationship("PostVote", back_populates="user")
106
favourites = db.relationship("RepoFavourite", back_populates="user")
107
commits = db.relationship("Commit", back_populates="owner")
109
posts = db.relationship("Post", back_populates="owner")
110
prs = db.relationship("PullRequest", back_populates="owner")
111
notifications = db.relationship("UserNotification", back_populates="user")
112
def __init__(self, username, password, email=None, display_name=None):
114
self.username = username
115
self.password_hashed = bcrypt.generate_password_hash(password, config.HASHING_ROUNDS).decode("utf-8")
116
self.email = ""
117
if email:
118
email_change_request = EmailChangeRequest(self, email)
119
db.session.add(email_change_request)
120
db.session.flush()
121
self.display_name = display_name
122
# Create the user's directory
124
if not os.path.exists(os.path.join(config.REPOS_PATH, username)):
125
os.makedirs(os.path.join(config.REPOS_PATH, username))
126
if not os.path.exists(os.path.join(config.USERDATA_PATH, username)):
127
os.makedirs(os.path.join(config.USERDATA_PATH, username))
128
avatar_name = random.choice(os.listdir(config.DEFAULT_AVATARS_PATH))
130
if os.path.join(config.DEFAULT_AVATARS_PATH, avatar_name).endswith(".svg"):
131
cairosvg.svg2png(url=os.path.join(config.DEFAULT_AVATARS_PATH, avatar_name),
132
write_to="/tmp/roundabout-avatar.png")
133
avatar = Image.open("/tmp/roundabout-avatar.png")
134
else:
135
avatar = Image.open(os.path.join(config.DEFAULT_AVATARS_PATH, avatar_name))
136
avatar.thumbnail(config.AVATAR_SIZE)
137
avatar.save(os.path.join(config.USERDATA_PATH, username, "avatar.png"))
138
# Create the configuration repo
140
config_repo = Repo(self, ".config", 0)
141
db.session.add(config_repo)
142
notification = Notification({"type": "welcome"})
143
db.session.add(notification)
144
db.session.commit()
145
user_notification = UserNotification(self, notification, 1)
147
db.session.add(user_notification)
148
db.session.flush()
149
celery_tasks.send_notification.apply_async(args=[user_notification.id])
150
class Repo(db.Model):
153
route = db.Column(db.String(98), unique=True, nullable=False, primary_key=True)
154
owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False)
155
name = db.Column(db.String(64), nullable=False)
156
owner = db.relationship("User", back_populates="repositories")
157
visibility = db.Column(db.SmallInteger(), nullable=False)
158
info = db.Column(db.Unicode(512), nullable=True)
159
url = db.Column(db.String(256), nullable=True)
160
creation_date = db.Column(db.DateTime, default=datetime.utcnow)
161
default_branch = db.Column(db.String(64), nullable=True, default="")
163
commits = db.relationship("Commit", back_populates="repo")
165
posts = db.relationship("Post", back_populates="repo")
166
repo_access = db.relationship("RepoAccess", back_populates="repo")
167
favourites = db.relationship("RepoFavourite", back_populates="repo")
168
heads = db.relationship("PullRequest", back_populates="head", foreign_keys="[PullRequest.head_route]")
169
bases = db.relationship("PullRequest", back_populates="base", foreign_keys="[PullRequest.base_route]")
170
has_site = db.Column(db.SmallInteger, nullable=False, default=0, # 0 means no site, 1 means it's got a site, 2 means it's the user's primary site
172
server_default="0") # (the one accessible at username.localhost)
173
site_branch = db.Column(db.String(64), nullable=True)
174
last_post_id = db.Column(db.Integer, nullable=False, default=0)
176
def __init__(self, owner, name, visibility):
178
self.route = f"/{owner.username}/{name}"
179
self.name = name
180
self.owner_name = owner.username
181
self.owner = owner
182
self.visibility = visibility
183
# Add the owner as an admin
185
repo_access = RepoAccess(owner, self, 2)
186
db.session.add(repo_access)
187
# Create the directory
189
if not os.path.exists(os.path.join(config.REPOS_PATH, self.owner_name, self.name)):
190
subprocess.run(["git", "init", self.name],
191
cwd=os.path.join(config.REPOS_PATH, self.owner_name))
192
class Commit(db.Model):
195
identifier = db.Column(db.String(227), unique=True, nullable=False, primary_key=True)
196
sha = db.Column(db.String(128), nullable=False)
197
repo_name = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False)
198
owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False)
199
owner_identity = db.Column(db.String(321))
200
receive_date = db.Column(db.DateTime, default=datetime.now)
201
author_date = db.Column(db.DateTime)
202
message = db.Column(db.UnicodeText)
203
repo = db.relationship("Repo", back_populates="commits")
204
owner = db.relationship("User", back_populates="commits")
205
def __init__(self, sha, owner, repo, date, message, owner_identity):
207
self.identifier = f"{repo.route}/{sha}"
208
self.sha = sha
209
self.repo_name = repo.route
210
self.repo = repo
211
self.owner_name = owner.username
212
self.owner = owner
213
self.author_date = datetime.fromtimestamp(int(date))
214
self.message = message
215
self.owner_identity = owner_identity
216
notification = Notification({"type": "commit", "repo": repo.route, "commit": sha})
218
db.session.add(notification)
219
db.session.flush() # save the notification to get the ID
220
# Send a notification to all users who have enabled commit notifications for this repo
222
for relationship in RepoFavourite.query.filter_by(repo_route=repo.route, notify_commit=True).all():
223
user = relationship.user
224
user_notification = UserNotification(user, notification, 1)
225
db.session.add(user_notification)
226
db.session.flush()
227
celery_tasks.send_notification.apply_async(args=[user_notification.id])
228
class Post(db.Model):
231
identifier = db.Column(db.String(109), unique=True, nullable=False, primary_key=True)
232
number = db.Column(db.Integer, nullable=False)
233
repo_name = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False)
234
owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False)
235
votes = db.relationship("PostVote", back_populates="post")
236
vote_sum = db.Column(db.Integer, nullable=False, default=0)
237
parent_id = db.Column(db.String(109), db.ForeignKey("post.identifier"), nullable=True)
239
root_id = db.Column(db.String(109), db.ForeignKey("post.identifier"), nullable=True)
240
state = db.Column(db.SmallInteger, nullable=True, default=1)
241
date = db.Column(db.DateTime, default=datetime.now)
243
last_updated = db.Column(db.DateTime, default=datetime.now)
244
subject = db.Column(db.Unicode(384))
245
message = db.Column(db.UnicodeText)
246
repo = db.relationship("Repo", back_populates="posts")
247
owner = db.relationship("User", back_populates="posts")
248
parent = db.relationship("Post", back_populates="children",
249
primaryjoin="Post.parent_id==Post.identifier",
250
foreign_keys="[Post.parent_id]", remote_side="Post.identifier")
251
root = db.relationship("Post",
252
primaryjoin="Post.root_id==Post.identifier",
253
foreign_keys="[Post.root_id]", remote_side="Post.identifier")
254
children = db.relationship("Post",
255
remote_side="Post.parent_id",
256
primaryjoin="Post.identifier==Post.parent_id",
257
foreign_keys="[Post.parent_id]")
258
def __init__(self, owner, repo, parent, subject, message):
260
self.identifier = f"{repo.route}/{repo.last_post_id}"
261
self.number = repo.last_post_id
262
self.repo_name = repo.route
263
self.repo = repo
264
self.owner_name = owner.username
265
self.owner = owner
266
self.subject = subject
267
self.message = message
268
self.parent = parent
269
if parent and parent.parent:
270
self.root = parent.parent
271
elif parent:
272
self.root = parent
273
else:
274
self.root = None
275
repo.last_post_id += 1
276
def update_date(self):
278
self.last_updated = datetime.now()
279
with db.session.no_autoflush:
280
if self.parent is not None:
281
self.parent.update_date()
282
class UserNotification(db.Model):
285
id = db.Column(db.Integer, primary_key=True)
286
user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False)
287
notification_id = db.Column(db.BigInteger, db.ForeignKey("notification.id"))
288
attention_level = db.Column(db.SmallInteger, nullable=False) # 0 is read
289
read_time = db.Column(db.DateTime, nullable=True)
290
user = db.relationship("User", back_populates="notifications")
292
notification = db.relationship("Notification", back_populates="notifications")
293
__table_args__ = (db.UniqueConstraint("user_username", "notification_id", name="_user_notification_uc"),)
295
def __init__(self, user, notification, level):
297
self.user_username = user.username
298
self.notification_id = notification.id
299
self.attention_level = level
300
def mark_read(self):
302
self.read_time = datetime.utcnow()
303
self.attention_level = 0
304
def mark_unread(self):
306
self.attention_level = 4
307
class UserFollow(db.Model):
310
id = db.Column(db.Integer, primary_key=True)
311
follower_username = db.Column(db.String(32), db.ForeignKey("user.username", ondelete="CASCADE"), nullable=False)
312
followed_username = db.Column(db.String(32), db.ForeignKey("user.username", ondelete="CASCADE"), nullable=False)
313
follower = db.relationship("User", back_populates="followers", foreign_keys=[follower_username])
315
followed = db.relationship("User", back_populates="follows", foreign_keys=[followed_username])
316
def __init__(self, follower_username, followed_username):
318
self.follower_username = follower_username
319
self.followed_username = followed_username
320
class Notification(db.Model):
323
id = db.Column(db.BigInteger, primary_key=True, autoincrement=True)
324
data = db.Column(db.dialects.postgresql.JSONB, nullable=False, default={})
325
notifications = db.relationship("UserNotification", back_populates="notification")
326
timestamp = db.Column(db.DateTime, nullable=False, default=datetime.now)
327
def __init__(self, json):
329
self.data = json
330
class PullRequest(db.Model):
333
id = db.Column(db.BigInteger, primary_key=True, autoincrement=True)
334
head_route = db.Column(db.String(98), db.ForeignKey("repo.route", ondelete="CASCADE"), nullable=False)
335
base_route = db.Column(db.String(98), db.ForeignKey("repo.route", ondelete="CASCADE"), nullable=False)
336
owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False)
337
state = db.Column(db.SmallInteger, nullable=False, default=0) # 0 pending, 1 merged, 2 rejected
338
head = db.relationship("Repo", back_populates="heads", foreign_keys=[head_route])
340
base = db.relationship("Repo", back_populates="bases", foreign_keys=[base_route])
341
head_branch = db.Column(db.String(64), nullable=False)
343
base_branch = db.Column(db.String(64), nullable=False)
344
owner = db.relationship("User", back_populates="prs")
346
timestamp = db.Column(db.DateTime, nullable=False, default=datetime.now)
347
def __init__(self, head, head_branch, base, base_branch, owner):
349
self.head = head
350
self.base = base
351
self.head_branch = head_branch
352
self.base_branch = base_branch
353
self.owner = owner
354
class EmailChangeRequest(db.Model):
356
id = db.Column(db.BigInteger, primary_key=True, autoincrement=True)
357
user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False)
358
new_email = db.Column(db.String(254), nullable=False)
359
code = db.Column(db.String(64), nullable=False)
360
expires_on = db.Column(db.DateTime, nullable=False)
361
user = db.relationship("User", back_populates="email_change_requests")
363
def __init__(self, user, new_email):
365
self.user = user
366
self.new_email = new_email
367
self.code = hex(secrets.randbits(256)).removeprefix("0x")
368
self.expires_on = datetime.now() + timedelta(days=1)
369