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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
|
"""Views regarding user collections."""
from uuid import UUID
from redis import Redis
from flask import jsonify, request, Response, Blueprint, current_app
from gn3.auth import db
from gn3.auth.db_utils import with_db_connection
from gn3.auth.authorisation.checks import require_json
from gn3.auth.authorisation.errors import NotFoundError
from gn3.auth.authentication.users import User, user_by_id
from gn3.auth.authentication.oauth2.resource_server import require_oauth
from .models import (
add_traits,
change_name,
remove_traits,
get_collection,
user_collections,
save_collections,
create_collection,
delete_collections as _delete_collections)
collections = Blueprint("collections", __name__)
@collections.route("/list")
@require_oauth("profile user")
def list_user_collections() -> Response:
"""Retrieve the user ids"""
with (require_oauth.acquire("profile user") as the_token,
Redis.from_url(current_app.config["REDIS_URI"],
decode_responses=True) as redisconn):
return jsonify(user_collections(redisconn, the_token.user))
@collections.route("/<uuid:anon_id>/list")
def list_anonymous_collections(anon_id: UUID) -> Response:
"""Fetch anonymous collections"""
with Redis.from_url(
current_app.config["REDIS_URI"], decode_responses=True) as redisconn:
def __list__(conn: db.DbConnection) -> tuple:
try:
_user = user_by_id(conn, anon_id)
current_app.logger.warning(
"Fetch collections for authenticated user using the "
"`list_user_collections()` endpoint.")
return tuple()
except NotFoundError as _nfe:
return user_collections(
redisconn, User(anon_id, "anon@ymous.user", "Anonymous User"))
return jsonify(with_db_connection(__list__))
@require_oauth("profile user")
def __new_collection_as_authenticated_user__(redisconn, name, traits):
"""Create a new collection as an authenticated user."""
with require_oauth.acquire("profile user") as token:
return create_collection(redisconn, token.user, name, traits)
def __new_collection_as_anonymous_user__(redisconn, name, traits):
"""Create a new collection as an anonymous user."""
return create_collection(redisconn,
User(UUID(request.json.get("anon_id")),
"anon@ymous.user",
"Anonymous User"),
name,
traits)
@collections.route("/new", methods=["POST"])
@require_json
def new_user_collection() -> Response:
"""Create a new collection."""
with (Redis.from_url(current_app.config["REDIS_URI"],
decode_responses=True) as redisconn):
traits = tuple(request.json.get("traits", tuple()))# type: ignore[union-attr]
name = request.json.get("name")# type: ignore[union-attr]
if bool(request.headers.get("Authorization")):
return jsonify(__new_collection_as_authenticated_user__(
redisconn, name, traits))
return jsonify(__new_collection_as_anonymous_user__(
redisconn, name, traits))
@collections.route("/<uuid:collection_id>/view", methods=["POST"])
@require_json
def view_collection(collection_id: UUID) -> Response:
"""View a particular collection"""
with (Redis.from_url(current_app.config["REDIS_URI"],
decode_responses=True) as redisconn):
if bool(request.headers.get("Authorization")):
with require_oauth.acquire("profile user") as token:
return jsonify(get_collection(redisconn,
token.user,
collection_id))
return jsonify(get_collection(
redisconn,
User(
UUID(request.json.get("anon_id")),#type: ignore[union-attr]
"anon@ymous.user",
"Anonymous User"),
collection_id))
@collections.route("/anonymous/import", methods=["POST"])
@require_json
@require_oauth("profile user")
def import_anonymous() -> Response:
"""Import anonymous collections."""
with (require_oauth.acquire("profile user") as token,
Redis.from_url(current_app.config["REDIS_URI"],
decode_responses=True) as redisconn):
anon_id = UUID(request.json.get("anon_id"))#type: ignore[union-attr]
anon_colls = user_collections(redisconn, User(
anon_id, "anon@ymous.user", "Anonymous User"))
save_collections(
redisconn,
token.user,
(user_collections(redisconn, token.user) +
anon_colls))
redisconn.hdel("collections", str(anon_id))
return jsonify({
"message": f"Import of {len(anon_colls)} was successful."
})
@collections.route("/anonymous/delete", methods=["POST"])
@require_json
@require_oauth("profile user")
def delete_anonymous() -> Response:
"""Delete anonymous collections."""
with (require_oauth.acquire("profile user") as _token,
Redis.from_url(current_app.config["REDIS_URI"],
decode_responses=True) as redisconn):
anon_id = UUID(request.json.get("anon_id"))#type: ignore[union-attr]
anon_colls = user_collections(redisconn, User(
anon_id, "anon@ymous.user", "Anonymous User"))
redisconn.hdel("collections", str(anon_id))
return jsonify({
"message": f"Deletion of {len(anon_colls)} was successful."
})
@collections.route("/delete", methods=["POST"])
@require_json
def delete_collections():
"""Delete specified collections."""
with (Redis.from_url(current_app.config["REDIS_URI"],
decode_responses=True) as redisconn):
coll_ids = tuple(UUID(cid) for cid in request.json["collection_ids"])
deleted = _delete_collections(
redisconn,
User(request.json["anon_id"], "anon@ymous.user", "Anonymous User"),
coll_ids)
if bool(request.headers.get("Authorization")):
with require_oauth.acquire("profile user") as token:
deleted = deleted + _delete_collections(
redisconn, token.user, coll_ids)
return jsonify({
"message": f"Deleted {len(deleted)} collections."})
@collections.route("/<uuid:collection_id>/traits/remove", methods=["POST"])
@require_json
def remove_traits_from_collection(collection_id: UUID) -> Response:
"""Remove specified traits from collection with ID `collection_id`."""
if len(request.json["traits"]) < 1:#type: ignore[index]
return jsonify({"message": "No trait to remove from collection."})
the_traits = tuple(request.json["traits"])#type: ignore[index]
with (Redis.from_url(current_app.config["REDIS_URI"],
decode_responses=True) as redisconn):
if not bool(request.headers.get("Authorization")):
coll = remove_traits(
redisconn,
User(request.json["anon_id"],#type: ignore[index]
"anon@ymous.user",
"Anonymous User"),
collection_id,
the_traits)
else:
with require_oauth.acquire("profile user") as token:
coll = remove_traits(
redisconn, token.user, collection_id, the_traits)
return jsonify({
"message": f"Deleted {len(the_traits)} traits from collection.",
"collection": coll
})
@collections.route("/<uuid:collection_id>/traits/add", methods=["POST"])
@require_json
def add_traits_to_collection(collection_id: UUID) -> Response:
"""Add specified traits to collection with ID `collection_id`."""
if len(request.json["traits"]) < 1:#type: ignore[index]
return jsonify({"message": "No trait to add to collection."})
the_traits = tuple(request.json["traits"])#type: ignore[index]
with (Redis.from_url(current_app.config["REDIS_URI"],
decode_responses=True) as redisconn):
if not bool(request.headers.get("Authorization")):
coll = add_traits(
redisconn,
User(request.json["anon_id"],#type: ignore[index]
"anon@ymous.user",
"Anonymous User"),
collection_id,
the_traits)
else:
with require_oauth.acquire("profile user") as token:
coll = add_traits(
redisconn, token.user, collection_id, the_traits)
return jsonify({
"message": f"Added {len(the_traits)} traits to collection.",
"collection": coll
})
@collections.route("/<uuid:collection_id>/rename", methods=["POST"])
@require_json
def rename_collection(collection_id: UUID) -> Response:
"""Rename the given collection"""
if not bool(request.json["new_name"]):#type: ignore[index]
return jsonify({"message": "No new name to change to."})
new_name = request.json["new_name"]#type: ignore[index]
with (Redis.from_url(current_app.config["REDIS_URI"],
decode_responses=True) as redisconn):
if not bool(request.headers.get("Authorization")):
coll = change_name(redisconn,
User(UUID(request.json["anon_id"]),#type: ignore[index]
"anon@ymous.user",
"Anonymous User"),
collection_id,
new_name)
else:
with require_oauth.acquire("profile user") as token:
coll = change_name(
redisconn, token.user, collection_id, new_name)
return jsonify({
"message": "Collection rename successful.",
"collection": coll
})
|
