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
|
"""The views/routes for the `gn3.auth.authorisation.groups` package."""
import uuid
import datetime
from functools import partial
from flask import request, jsonify, Response, Blueprint, current_app
from gn3.auth import db
from gn3.auth.dictify import dictify
from gn3.auth.db_utils import with_db_connection
from .models import (
user_group, all_groups, join_requests, GroupCreationError,
accept_reject_join_request, group_users as _group_users,
create_group as _create_group)
from ..errors import AuthorisationError
from ...authentication.users import User
from ...authentication.oauth2.resource_server import require_oauth
groups = Blueprint("groups", __name__)
@groups.route("/list", methods=["GET"])
@require_oauth("profile group")
def list_groups():
"""Return the list of groups that exist."""
with db.connection(current_app.config["AUTH_DB"]) as conn:
the_groups = all_groups(conn)
return jsonify(the_groups.maybe(
[], lambda grps: [dictify(grp) for grp in grps]))
@groups.route("/create", methods=["POST"])
@require_oauth("profile group")
def create_group():
"""Create a new group."""
with require_oauth.acquire("profile group") as the_token:
group_name=request.form.get("group_name", "").strip()
if not bool(group_name):
raise GroupCreationError("Could not create the group.")
db_uri = current_app.config["AUTH_DB"]
with db.connection(db_uri) as conn:
user = the_token.user
new_group = _create_group(
conn, group_name, user, request.form.get("group_description"))
return jsonify({
**dictify(new_group), "group_leader": dictify(user)
})
@groups.route("/members/<uuid:group_id>", methods=["GET"])
@require_oauth("profile group")
def group_members(group_id: uuid.UUID) -> Response:
"""Retrieve all the members of a group."""
with require_oauth.acquire("profile group") as the_token:# pylint: disable=[unused-variable]
db_uri = current_app.config["AUTH_DB"]
## Check that user has appropriate privileges and remove the pylint disable above
with db.connection(db_uri) as conn:
return jsonify(tuple(
dictify(user) for user in _group_users(conn, group_id)))
@groups.route("/requests/join/<uuid:group_id>", methods=["POST"])
@require_oauth("profile group")
def request_to_join(group_id: uuid.UUID) -> Response:
"""Request to join a group."""
def __request__(conn: db.DbConnection, user: User, group_id: uuid.UUID,
message: str):
with db.cursor(conn) as cursor:
group = user_group(cursor, user).maybe(# type: ignore[misc]
False, lambda grp: grp)# type: ignore[arg-type]
if group:
error = AuthorisationError(
"You cannot request to join a new group while being a "
"member of an existing group.")
error.error_code = 400
raise error
request_id = uuid.uuid4()
cursor.execute(
"INSERT INTO group_join_requests VALUES "
"(:request_id, :group_id, :user_id, :ts, :status, :msg)",
{
"request_id": str(request_id),
"group_id": str(group_id),
"user_id": str(user.user_id),
"ts": datetime.datetime.now().timestamp(),
"status": "PENDING",
"msg": message
})
return {
"request_id": request_id,
"message": "Successfully sent the join request."
}
with require_oauth.acquire("profile group") as the_token:
form = request.form
results = with_db_connection(partial(
__request__, user=the_token.user, group_id=group_id, message=form.get(
"message", "I hereby request that you add me to your group.")))
return jsonify(results)
@groups.route("/requests/join/list", methods=["GET"])
@require_oauth("profile group")
def list_join_requests() -> Response:
"""List the pending join requests."""
with require_oauth.acquire("profile group") as the_token:
return jsonify(with_db_connection(partial(
join_requests, user=the_token.user)))
@groups.route("/requests/join/accept", methods=["POST"])
@require_oauth("profile group")
def accept_join_requests() -> Response:
"""Accept a join request."""
with require_oauth.acquire("profile group") as the_token:
form = request.form
request_id = uuid.UUID(form.get("request_id"))
return jsonify(with_db_connection(partial(
accept_reject_join_request, request_id=request_id,
user=the_token.user, status="ACCEPTED")))
@groups.route("/requests/join/reject", methods=["POST"])
@require_oauth("profile group")
def reject_join_requests() -> Response:
"""Reject a join request."""
with require_oauth.acquire("profile group") as the_token:
form = request.form
request_id = uuid.UUID(form.get("request_id"))
return jsonify(with_db_connection(partial(
accept_reject_join_request, request_id=request_id,
user=the_token.user, status="REJECTED")))
|
