Merge pull request #94 from kbaseapps/add-ruff-1

UIP-49 Add ruff - first pass

.github/workflows/run_tests.yml

@@ -27,6 +27,7 @@ jobs:
         shell: bash -l {0}
         run: |
           cat requirements.txt | sed -e '/^\s*#.*$/d' -e '/^\s*$/d' | xargs -n 1 pip install
+          cat requirements-dev.txt | sed -e '/^\s*#.*$/d' -e '/^\s*$/d' | xargs -n 1 pip install
 
       - name: Run tests
         shell: bash -l {0}

.gitignore

@@ -2,4 +2,8 @@ test_local
 sdk.cfg
 .DS_Store
 .vscode
+__pycache__
 .env
+.coverage
+.coveragerc
+default.profraw

lib/NarrativeService/DynamicServiceCache.py

@@ -1,10 +1,7 @@
-# -*- coding: utf-8 -*-
 
 import time
-from .baseclient import (
-    BaseClient,
-    ServerError
-)
+
+from .baseclient import BaseClient, ServerError
 
 
 class DynamicServiceClient:
@@ -45,11 +42,11 @@ def call_method(self, method: str, params_array: list) -> list:
 
     def _lookup_url(self):
         bc = BaseClient(url=self.sw_url, lookup_url=False)
-        self.cached_url = bc.call_method('ServiceWizard.get_service_status',
-                                         [{'module_name': self.module_name,
-                                           'version': self.service_ver}])['url']
+        self.cached_url = bc.call_method("ServiceWizard.get_service_status",
+                                         [{"module_name": self.module_name,
+                                           "version": self.service_ver}])["url"]
         self.last_refresh_time = time.time()
 
     def _call(self, method, params_array, token):
         bc = BaseClient(url=self.cached_url, token=token, lookup_url=False)
-        return bc.call_method(self.module_name + '.' + method, params_array)
+        return bc.call_method(self.module_name + "." + method, params_array)

lib/NarrativeService/NarrativeListUtils.py

@@ -13,7 +13,7 @@
 #     8 usermeta metadata
 
 
-class NarrativeInfoCache(object):
+class NarrativeInfoCache:
 
     def __init__(self, cache_size):
         self.cache = pylru.lrucache(int(cache_size))
@@ -25,7 +25,7 @@ def check_cache_size(self):
         return len(self.cache)
 
     def get_info_list(self, ws_lookup_table, wsClient):
-        '''
+        """
             Given a set of WS info, lookup the corresponding Narrative
             information.
 
@@ -38,11 +38,11 @@ def get_info_list(self, ws_lookup_table, wsClient):
                  },
                  ...
                  ]
-        '''
+        """
         # search the cache and extract what we have, mark what was missed
         res = self._search_cache(ws_lookup_table)
-        items = res['items']
-        missed_items = res['missed']
+        items = res["items"]
+        missed_items = res["missed"]
 
         # for objects that were missed, we have to go out and fetch that
         # narrative object info
@@ -59,43 +59,43 @@ def _search_cache(self, ws_lookup_table):
         for ws_info in ws_lookup_table.values():
             key = self._get_cache_key(ws_info)
             if key in self.cache:
-                items.append({'ws': ws_info, 'nar': self.cache[key]})
+                items.append({"ws": ws_info, "nar": self.cache[key]})
             else:
                 missed.append(ws_info)
-        return {'items': items, 'missed': missed}
+        return {"items": items, "missed": missed}
 
     def _fetch_objects_and_cache(self, ws_list, full_ws_lookup_table, wsClient):
-        ''' Fetches narrative objects (if possible) for everything in ws_list '''
+        """ Fetches narrative objects (if possible) for everything in ws_list """
         obj_ref_list = []
         for ws_info in ws_list:
-            if 'narrative' not in ws_info[8]:
+            if "narrative" not in ws_info[8]:
                 continue
-            if not ws_info[8]['narrative'].isdigit() or not int(ws_info[8]['narrative']) > 0:
+            if not ws_info[8]["narrative"].isdigit() or not int(ws_info[8]["narrative"]) > 0:
                 continue
-            ref = str(ws_info[0]) + '/' + str(ws_info[8]['narrative'])
-            obj_ref_list.append({'ref': ref})
+            ref = str(ws_info[0]) + "/" + str(ws_info[8]["narrative"])
+            obj_ref_list.append({"ref": ref})
 
         if len(obj_ref_list) == 0:
             return []
 
         # ignore errors
-        get_obj_params = {'objects': obj_ref_list, 'includeMetadata': 1, 'ignoreErrors': 1}
-        narrative_list = wsClient.get_object_info3(get_obj_params)['infos']
+        get_obj_params = {"objects": obj_ref_list, "includeMetadata": 1, "ignoreErrors": 1}
+        narrative_list = wsClient.get_object_info3(get_obj_params)["infos"]
 
         items = []
         for nar in narrative_list:
             if nar:
                 ws_info = full_ws_lookup_table[nar[6]]
-                items.append({'ws': ws_info, 'nar': nar})
+                items.append({"ws": ws_info, "nar": nar})
                 self.cache[self._get_cache_key(ws_info)] = nar
         return items
 
     def _get_cache_key(self, ws_info):
-        return str(ws_info[0]) + '__' + str(ws_info[3])
+        return str(ws_info[0]) + "__" + str(ws_info[3])
 
 
 
-class NarrativeListUtils(object):
+class NarrativeListUtils:
 
     def __init__(self, cache_size):
         self.narrativeInfo = NarrativeInfoCache(cache_size)
@@ -106,7 +106,7 @@ def list_public_narratives(self, wsClient):
         ws_list = wsClient.list_workspace_info({})
         ws_global_list = []
         for ws_info in ws_list:
-            if ws_info[6] == 'r':  # indicates that this ws is globally readable
+            if ws_info[6] == "r":  # indicates that this ws is globally readable
                 ws_global_list.append(ws_info)
         # build a ws_list lookup table
         ws_lookup_table = self._build_ws_lookup_table(ws_global_list)
@@ -116,7 +116,7 @@ def list_public_narratives(self, wsClient):
 
     def list_my_narratives(self, my_user_id, wsClient):
         # get all the workspaces marked as narratorials
-        ws_list = wsClient.list_workspace_info({'owners': [my_user_id]})
+        ws_list = wsClient.list_workspace_info({"owners": [my_user_id]})
         # build a ws_list lookup table
         ws_lookup_table = self._build_ws_lookup_table(ws_list)
         # based on the WS lookup table, lookup the narratives
@@ -130,7 +130,7 @@ def list_shared_narratives(self, my_user_id, wsClient):
         for ws_info in ws_list:
             if ws_info[2] == my_user_id:
                 continue
-            if ws_info[5] == 'n':  # indicates that this ws is globally readable
+            if ws_info[5] == "n":  # indicates that this ws is globally readable
                 continue
             ws_shared_list.append(ws_info)
         # build a ws_list lookup table
@@ -141,39 +141,39 @@ def list_shared_narratives(self, my_user_id, wsClient):
 
     def list_narratorials(self, wsClient):
         # get all the workspaces marked as narratorials
-        ws_list = wsClient.list_workspace_info({'meta': {'narratorial': '1'}})
+        ws_list = wsClient.list_workspace_info({"meta": {"narratorial": "1"}})
         # build a ws_list lookup table
         ws_lookup_table = self._build_ws_lookup_table(ws_list)
         # based on the WS lookup table, lookup the narratives
         return self.narrativeInfo.get_info_list(ws_lookup_table, wsClient)
 
 
     def _build_ws_lookup_table(self, ws_list):
-        ''' builds a lookup table, skips anything without a 'narrative' metadata field set '''
+        """ builds a lookup table, skips anything without a 'narrative' metadata field set """
         ws_lookup_table = {}
         for ws_info in ws_list:
-            if 'narrative' in ws_info[8]:
-                if ws_info[8]['narrative'].isdigit() and int(ws_info[8]['narrative']) > 0:
+            if "narrative" in ws_info[8]:
+                if ws_info[8]["narrative"].isdigit() and int(ws_info[8]["narrative"]) > 0:
                     ws_lookup_table[ws_info[0]] = ws_info
         return ws_lookup_table
 
 
-class NarratorialUtils(object):
+class NarratorialUtils:
 
     def __init__(self):
         pass
 
     def _get_workspace_identity(self, wsid):
         if str(wsid).isdigit():
-            return {'id': int(wsid)}
+            return {"id": int(wsid)}
         else:
-            return {'workspace': str(wsid)}
+            return {"workspace": str(wsid)}
 
     def set_narratorial(self, wsid, description, wsClient):
         wsi = self._get_workspace_identity(wsid)
-        wsClient.alter_workspace_metadata({'wsi': wsi, 'new': {'narratorial': '1'}})
-        wsClient.alter_workspace_metadata({'wsi': wsi, 'new': {'narratorial_description': description}})
+        wsClient.alter_workspace_metadata({"wsi": wsi, "new": {"narratorial": "1"}})
+        wsClient.alter_workspace_metadata({"wsi": wsi, "new": {"narratorial_description": description}})
 
     def remove_narratorial(self, wsid, wsClient):
         wsi = self._get_workspace_identity(wsid)
-        wsClient.alter_workspace_metadata({'wsi': wsi, 'remove': ['narratorial', 'narratorial_description']})
+        wsClient.alter_workspace_metadata({"wsi": wsi, "remove": ["narratorial", "narratorial_description"]})