Source, Obj, Candidate, and Permissions Handler Refactor #1: Rename O…

…wned -> Readable (skyportal#1441)

* update baselayer

* rename owned -> readable

* linting

skyportal/handlers/api/annotation.py

@@ -27,7 +27,7 @@ def get(self, annotation_id):
               application/json:
                 schema: Error
         """
-        annotation = Annotation.get_if_owned_by(annotation_id, self.current_user)
+        annotation = Annotation.get_if_readable_by(annotation_id, self.current_user)
         if annotation is None:
             return self.error('Invalid annotation ID.')
         return self.success(data=annotation)
@@ -204,7 +204,7 @@ def put(self, annotation_id):
               application/json:
                 schema: Error
         """
-        a = Annotation.get_if_owned_by(annotation_id, self.current_user)
+        a = Annotation.get_if_readable_by(annotation_id, self.current_user)
         if a is None:
             return self.error('Invalid annotation ID.')
 
@@ -219,7 +219,7 @@ def put(self, annotation_id):
             return self.error(f'Invalid/missing parameters: {e.normalized_messages()}')
         DBSession().flush()
         if group_ids is not None:
-            a = Annotation.get_if_owned_by(annotation_id, self.current_user)
+            a = Annotation.get_if_readable_by(annotation_id, self.current_user)
             groups = Group.query.filter(Group.id.in_(group_ids)).all()
             if not groups:
                 return self.error(

skyportal/handlers/api/candidate.py

@@ -254,7 +254,7 @@ def get(self, obj_id=None):
                     .joinedload(Obj.photometry)
                     .joinedload(Photometry.instrument)
                 )
-            c = Candidate.get_obj_if_owned_by(
+            c = Candidate.get_obj_if_readable_by(
                 obj_id, self.current_user, options=query_options,
             )
             if c is None:
@@ -286,12 +286,16 @@ def get(self, obj_id=None):
             candidate_info["filter_ids"] = filter_ids
             candidate_info["passing_alerts"] = passing_alerts
             candidate_info["comments"] = sorted(
-                [cmt.to_dict() for cmt in c.get_comments_owned_by(self.current_user)],
+                [
+                    cmt.to_dict()
+                    for cmt in c.get_comments_readable_by(self.current_user)
+                ],
                 key=lambda x: x["created_at"],
                 reverse=True,
             )
             candidate_info["annotations"] = sorted(
-                c.get_annotations_owned_by(self.current_user), key=lambda x: x.origin,
+                c.get_annotations_readable_by(self.current_user),
+                key=lambda x: x.origin,
             )
             candidate_info["is_source"] = len(c.sources) > 0
             if candidate_info["is_source"]:
@@ -304,7 +308,7 @@ def get(self, obj_id=None):
                     .filter(Group.id.in_(user_accessible_group_ids))
                     .all()
                 )
-                candidate_info["classifications"] = c.get_classifications_owned_by(
+                candidate_info["classifications"] = c.get_classifications_readable_by(
                     self.current_user
                 )
             candidate_info["last_detected"] = c.last_detected
@@ -537,7 +541,7 @@ def get(self, obj_id=None):
                         .filter(Group.id.in_(user_accessible_group_ids))
                         .all()
                     )
-                    obj.classifications = obj.get_classifications_owned_by(
+                    obj.classifications = obj.get_classifications_readable_by(
                         self.current_user
                     )
                 obj.passing_group_ids = [
@@ -560,13 +564,13 @@ def get(self, obj_id=None):
                 candidate_list[-1]["comments"] = sorted(
                     [
                         cmt.to_dict()
-                        for cmt in obj.get_comments_owned_by(self.current_user)
+                        for cmt in obj.get_comments_readable_by(self.current_user)
                     ],
                     key=lambda x: x["created_at"],
                     reverse=True,
                 )
                 candidate_list[-1]["annotations"] = sorted(
-                    obj.get_annotations_owned_by(self.current_user),
+                    obj.get_annotations_readable_by(self.current_user),
                     key=lambda x: x.origin,
                 )
                 candidate_list[-1]["last_detected"] = obj.last_detected

skyportal/handlers/api/classification.py

@@ -28,7 +28,7 @@ def get(self, classification_id):
               application/json:
                 schema: Error
         """
-        classification = Classification.get_if_owned_by(
+        classification = Classification.get_if_readable_by(
             classification_id, self.current_user
         )
         if classification is None:
@@ -94,7 +94,7 @@ def post(self):
         data = self.get_json()
         obj_id = data['obj_id']
         # Ensure user/token has access to parent source
-        source = Source.get_obj_if_owned_by(obj_id, self.current_user)
+        source = Source.get_obj_if_readable_by(obj_id, self.current_user)
         if source is None:
             return self.error("Invalid source.")
         user_group_ids = [g.id for g in self.current_user.groups]
@@ -214,7 +214,7 @@ def put(self, classification_id):
               application/json:
                 schema: Error
         """
-        c = Classification.get_if_owned_by(classification_id, self.current_user)
+        c = Classification.get_if_readable_by(classification_id, self.current_user)
         if c is None:
             return self.error('Invalid classification ID.')
 
@@ -231,7 +231,7 @@ def put(self, classification_id):
             )
         DBSession().flush()
         if group_ids is not None:
-            c = Classification.get_if_owned_by(classification_id, self.current_user)
+            c = Classification.get_if_readable_by(classification_id, self.current_user)
             groups = Group.query.filter(Group.id.in_(group_ids)).all()
             if not groups:
                 return self.error(

skyportal/handlers/api/comment.py

@@ -28,7 +28,7 @@ def get(self, comment_id):
               application/json:
                 schema: Error
         """
-        comment = Comment.get_if_owned_by(comment_id, self.current_user)
+        comment = Comment.get_if_readable_by(comment_id, self.current_user)
         if comment is None:
             return self.error('Invalid comment ID.')
         return self.success(data=comment)
@@ -92,7 +92,7 @@ def post(self):
         comment_text = data.get("text")
 
         # Ensure user/token has access to parent source
-        _ = Source.get_obj_if_owned_by(obj_id, self.current_user)
+        _ = Source.get_obj_if_readable_by(obj_id, self.current_user)
         user_accessible_group_ids = [g.id for g in self.current_user.accessible_groups]
         user_accessible_filter_ids = [
             filtr.id
@@ -202,7 +202,7 @@ def put(self, comment_id):
               application/json:
                 schema: Error
         """
-        c = Comment.get_if_owned_by(comment_id, self.current_user)
+        c = Comment.get_if_readable_by(comment_id, self.current_user)
         if c is None:
             return self.error('Invalid comment ID.')
 
@@ -233,7 +233,7 @@ def put(self, comment_id):
 
         DBSession().flush()
         if group_ids is not None:
-            c = Comment.get_if_owned_by(comment_id, self.current_user)
+            c = Comment.get_if_readable_by(comment_id, self.current_user)
             groups = Group.query.filter(Group.id.in_(group_ids)).all()
             if not groups:
                 return self.error(
@@ -328,7 +328,7 @@ def get(self, comment_id):
         """
         download = strtobool(self.get_query_argument('download', "True").lower())
 
-        comment = Comment.get_if_owned_by(comment_id, self.current_user)
+        comment = Comment.get_if_readable_by(comment_id, self.current_user)
         if comment is None:
             return self.error('Invalid comment ID.')
 

skyportal/handlers/api/followup_request.py

@@ -145,7 +145,7 @@ def post(self):
             return self.error('Object is already assigned to this run.')
 
         assignment = ClassicalAssignment(**data)
-        source = Source.get_obj_if_owned_by(assignment.obj_id, self.current_user)
+        source = Source.get_obj_if_readable_by(assignment.obj_id, self.current_user)
 
         if source is None:
             return self.error(f'Invalid obj_id: "{assignment.obj_id}"')
@@ -341,7 +341,7 @@ def post(self):
                               description: New follow-up request ID
         """
         data = self.get_json()
-        _ = Source.get_obj_if_owned_by(data["obj_id"], self.current_user)
+        _ = Source.get_obj_if_readable_by(data["obj_id"], self.current_user)
 
         try:
             data = FollowupRequestPost.load(data)
@@ -429,7 +429,7 @@ def put(self, request_id):
                 schema: Error
         """
 
-        followup_request = FollowupRequest.get_if_owned_by(
+        followup_request = FollowupRequest.get_if_readable_by(
             request_id, self.current_user
         )
 
@@ -497,7 +497,7 @@ def delete(self, request_id):
               application/json:
                 schema: Success
         """
-        followup_request = FollowupRequest.get_if_owned_by(
+        followup_request = FollowupRequest.get_if_readable_by(
             request_id, self.current_user
         )
         if not (

skyportal/handlers/api/internal/plot.py

@@ -37,7 +37,7 @@ def get(self, obj_id):
 
 class AirmassHandler(BaseHandler):
     def calculate_airmass(self, obj, telescope, sunset, sunrise):
-        permission_check = Source.get_obj_if_owned_by(obj.id, self.current_user)
+        permission_check = Source.get_obj_if_readable_by(obj.id, self.current_user)
         if permission_check is None:
             return self.error('Invalid assignment id.')
 
@@ -84,7 +84,7 @@ def get(self, obj_id, telescope_id):
         else:
             time = ap_time.Time.now()
 
-        obj = Source.get_obj_if_owned_by(obj_id, self.current_user)
+        obj = Source.get_obj_if_readable_by(obj_id, self.current_user)
         if obj is None:
             return self.error('Invalid assignment id.')
 

skyportal/handlers/api/internal/recent_sources.py

@@ -55,7 +55,7 @@ def get(self):
                 recency_index = sources_seen[obj_id]
                 sources_seen[obj_id] += 1
 
-            s = Source.get_obj_if_owned_by(  # Returns Source.obj
+            s = Source.get_obj_if_readable_by(  # Returns Source.obj
                 obj_id,
                 self.current_user,
                 options=[joinedload(Source.obj).joinedload(Obj.thumbnails)],

skyportal/handlers/api/internal/source_views.py

@@ -49,7 +49,7 @@ def get(self):
         )
         sources = []
         for view, obj_id in query_results:
-            s = Source.get_obj_if_owned_by(  # Returns Source.obj
+            s = Source.get_obj_if_readable_by(  # Returns Source.obj
                 obj_id,
                 self.current_user,
                 options=[joinedload(Source.obj).joinedload(Obj.thumbnails)],
@@ -71,7 +71,7 @@ def get(self):
     @tornado.web.authenticated
     def post(self, obj_id):
         # Ensure user has access to source
-        Source.get_obj_if_owned_by(obj_id, self.current_user)
+        Source.get_obj_if_readable_by(obj_id, self.current_user)
         # This endpoint will only be hit by front-end, so this will never be a token
         register_source_view(
             obj_id=obj_id,

skyportal/handlers/api/internal/token.py

@@ -74,7 +74,7 @@ def delete(self, token_id):
               application/json:
                 schema: Error
         """
-        token = Token.get_if_owned_by(token_id, self.current_user)
+        token = Token.get_if_readable_by(token_id, self.current_user)
         if token is not None:
             DBSession.delete(token)
             DBSession.commit()

skyportal/handlers/api/observingrun.py

@@ -128,7 +128,7 @@ def get(self, run_id=None):
             assignments = []
             for a in run.assignments:
                 try:
-                    obj = Obj.get_if_owned_by(a.obj.id, self.current_user)
+                    obj = Obj.get_if_readable_by(a.obj.id, self.current_user)
                 except AccessError:
                     continue
                 if obj is not None:

skyportal/handlers/api/photometry.py

@@ -780,7 +780,7 @@ def put(self):
     def get(self, photometry_id):
         # The full docstring/API spec is below as an f-string
 
-        phot = Photometry.get_if_owned_by(photometry_id, self.current_user)
+        phot = Photometry.get_if_readable_by(photometry_id, self.current_user)
         if phot is None:
             return self.error('Invalid photometry ID')
 
@@ -819,7 +819,7 @@ def patch(self, photometry_id):
                 schema: Error
         """
 
-        photometry = Photometry.get_if_owned_by(photometry_id, self.current_user)
+        photometry = Photometry.get_if_readable_by(photometry_id, self.current_user)
         if not photometry.is_modifiable_by(self.associated_user_object):
             return self.error(
                 f'Cannot delete photometry point that is owned by {photometry.owner}.'
@@ -885,7 +885,7 @@ def delete(self, photometry_id):
               application/json:
                 schema: Error
         """
-        photometry = Photometry.get_if_owned_by(photometry_id, self.current_user)
+        photometry = Photometry.get_if_readable_by(photometry_id, self.current_user)
         if not photometry.is_modifiable_by(self.associated_user_object):
             return self.error(
                 f'Cannot delete photometry point that is owned by {photometry.owner}.'
@@ -905,7 +905,7 @@ def get(self, obj_id):
         obj = Obj.query.get(obj_id)
         if obj is None:
             return self.error('Invalid object id.')
-        photometry = Obj.get_photometry_owned_by_user(obj_id, self.current_user)
+        photometry = Obj.get_photometry_readable_by_user(obj_id, self.current_user)
         format = self.get_query_argument('format', 'mag')
         outsys = self.get_query_argument('magsys', 'ab')
         return self.success(
@@ -937,7 +937,7 @@ def delete(self, upload_id):
         """
         # Permissions check:
         phot_id = Photometry.query.filter(Photometry.upload_id == upload_id).first().id
-        _ = Photometry.get_if_owned_by(phot_id, self.current_user)
+        _ = Photometry.get_if_readable_by(phot_id, self.current_user)
 
         n_deleted = (
             DBSession()