Python script, ASCII text executable
1
import time
2
import os
3
import config
4
import email_send
5
from celery import shared_task
6
from app import db
7
from misc_utils import *
8
from models import *
9
from smtplib import SMTP
10
@shared_task(ignore_result=False)
13
def send_notification(notification_id, users, level):
14
notification = db.session.get(Notification, notification_id)
15
for user in users:
17
db.session.add(UserNotification(db.session.get(User, user), notification, level))
18
with (SMTP(config.MAIL_SERVER) as mail):
20
if notification.data.get("type") == "welcome":
21
message = ("Subject:Welcome"
22
+ email_send.render_email_template("welcome.html", username=user))
23
mail.sendmail(
24
config.NOTIFICATION_EMAIL,
25
db.session.get(User, user).email,
26
message,
27
)
28
db.session.commit()
29
return 0 # notification sent successfully
31
@shared_task(ignore_result=False)
34
def merge_heads(head_route, head_branch, base_route, base_branch):
35
server_repo_location = os.path.join(config.REPOS_PATH, base_route.lstrip("/"))
36
if not os.path.isdir(server_repo_location):
37
raise FileNotFoundError(f"Repo {server_repo_location} not found, cannot merge.")
38
if base_route == head_route:
40
git_command(server_repo_location, b"", "checkout", f"{base_branch}")
41
out, err = git_command(server_repo_location, b"", "merge", "--no-ff", f"heads/{head_branch}", return_err=True)
42
return out, err
44
remote_url = os.path.join(config.BASE_DOMAIN, "git", base_route.lstrip("/"))
46
git_command(server_repo_location, b"", "remote", "add", "NEW", remote_url)
48
git_command(server_repo_location, b"", "remote", "update")
49
git_command(server_repo_location, b"", "checkout", f"{base_branch}")
50
git_command(server_repo_location, b"", "merge", "--allow-unrelated-histories", f"NEW/{head_branch}")
51
git_command(server_repo_location, b"", "remote", "rm", "NEW")
52