aboutsummaryrefslogtreecommitdiff
path: root/gn3/auth/authorisation
diff options
context:
space:
mode:
authorFrederick Muriuki Muriithi2023-05-18 14:05:15 +0300
committerFrederick Muriuki Muriithi2023-05-18 14:05:15 +0300
commitc43450502d1645a2df394fd7c0f7b655e675ab25 (patch)
tree4737347f801de8756946a40c8e8e6e58fa72a249 /gn3/auth/authorisation
parentf21f39ef1926dc24765626f3879cc5f81a05faad (diff)
downloadgenenetwork3-c43450502d1645a2df394fd7c0f7b655e675ab25.tar.gz
auth: user collections: Add and/or remove traits.
Diffstat (limited to 'gn3/auth/authorisation')
-rw-r--r--gn3/auth/authorisation/users/collections/models.py75
-rw-r--r--gn3/auth/authorisation/users/collections/views.py63
2 files changed, 137 insertions, 1 deletions
diff --git a/gn3/auth/authorisation/users/collections/models.py b/gn3/auth/authorisation/users/collections/models.py
index eaee9af..28223cb 100644
--- a/gn3/auth/authorisation/users/collections/models.py
+++ b/gn3/auth/authorisation/users/collections/models.py
@@ -146,6 +146,26 @@ def get_collection(rconn: Redis, user: User, collection_id: UUID) -> dict:
raise err
return colls[0]
+def __raise_if_collections_empty__(user: User, collections: tuple[dict, ...]):
+ """Raise an exception if no collections are found for `user`."""
+ if len(collections) < 1:
+ raise NotFoundError(f"No collections found for user `{user.user_id}`")
+
+def __raise_if_not_single_collection__(
+ user: User, collection_id: UUID, collections: tuple[dict, ...]):
+ """
+ Raise an exception there is zero, or more than one collection for `user`.
+ """
+ if len(collections) == 0:
+ raise NotFoundError(f"No collections found for user `{user.user_id}` "
+ f"with ID `{collection_id}`.")
+ if len(collections) > 1:
+ err = InvalidData(
+ "More than one collection was found having the ID "
+ f"`{collection_id}` for user with ID `{user.user_id}`.")
+ err.error_code = 513
+ raise err
+
def delete_collections(rconn: Redis,
user: User,
collection_ids: tuple[UUID, ...]) -> tuple[dict, ...]:
@@ -159,3 +179,58 @@ def delete_collections(rconn: Redis,
user,
tuple(coll for coll in ucolls if coll["id"] not in collection_ids))
return tuple(coll for coll in ucolls if coll["id"] in collection_ids)
+
+def add_traits(rconn: Redis,
+ user: User,
+ collection_id: UUID,
+ traits: tuple[str, ...]) -> dict:
+ """
+ Add `traits` to the `user` collection identified by `collection_id`.
+
+ Returns: The collection with the new traits added.
+ """
+ ucolls = user_collections(rconn, user)
+ __raise_if_collections_empty__(user, ucolls)
+
+ mod_col = tuple(coll for coll in ucolls if coll["id"] == collection_id)
+ __raise_if_not_single_collection__(user, collection_id, mod_col)
+ new_members = tuple(set(tuple(mod_col[0]["members"]) + traits))
+ new_coll = {
+ **mod_col[0],
+ "members": new_members,
+ "num_members": len(new_members)
+ }
+ save_collections(
+ rconn,
+ user,
+ (tuple(coll for coll in ucolls if coll["id"] != collection_id) +
+ (new_coll,)))
+ return new_coll
+
+def remove_traits(rconn: Redis,
+ user: User,
+ collection_id: UUID,
+ traits: tuple[str, ...]) -> dict:
+ """
+ Remove `traits` from the `user` collection identified by `collection_id`.
+
+ Returns: The collection with the specified `traits` removed.
+ """
+ ucolls = user_collections(rconn, user)
+ __raise_if_collections_empty__(user, ucolls)
+
+ mod_col = tuple(coll for coll in ucolls if coll["id"] == collection_id)
+ __raise_if_not_single_collection__(user, collection_id, mod_col)
+ new_members = tuple(
+ trait for trait in mod_col[0]["members"] if trait not in traits)
+ new_coll = {
+ **mod_col[0],
+ "members": new_members,
+ "num_members": len(new_members)
+ }
+ save_collections(
+ rconn,
+ user,
+ (tuple(coll for coll in ucolls if coll["id"] != collection_id) +
+ (new_coll,)))
+ return new_coll
diff --git a/gn3/auth/authorisation/users/collections/views.py b/gn3/auth/authorisation/users/collections/views.py
index 419b460..c8d470d 100644
--- a/gn3/auth/authorisation/users/collections/views.py
+++ b/gn3/auth/authorisation/users/collections/views.py
@@ -13,7 +13,12 @@ from gn3.auth.authentication.users import User, user_by_id
from gn3.auth.authentication.oauth2.resource_server import require_oauth
from .models import (
- get_collection, user_collections, save_collections, create_collection,
+ add_traits,
+ remove_traits,
+ get_collection,
+ user_collections,
+ save_collections,
+ create_collection,
delete_collections as _delete_collections)
collections = Blueprint("collections", __name__)
@@ -148,3 +153,59 @@ def delete_collections():
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
+ })