Support more filters when listing registered directories.

src/sewerrat/list_registered_directories.py

@@ -1,8 +1,9 @@
-from typing import Optional, Union
+from typing import Optional, Union, List, Dict
 import requests
+import urllib
 
 
-def list_registered_directories(url: str, user: Optional[Union[str, bool]] = None):
+def list_registered_directories(url: str, user: Optional[Union[str, bool]] = None, contains: Optional[str] = None, prefix: Optional[str] = None) -> List[Dict]:
     """
     List all registered directories in the SewerRat instance.
 
@@ -11,23 +12,39 @@ def list_registered_directories(url: str, user: Optional[Union[str, bool]] = Non
             URL to the SewerRat REST API.
 
         user:
-            Name of a user, used to filter the returned directories based on
-            the user who registered them. Alternatively True, to automatically
-            use the name of the current user.
+            Name of a user. If not None, this is used to filter the returned
+            directories based on the user who registered them. Alternatively
+            True, to automatically use the name of the current user.
+
+        contains:
+            String containing an absolute path. If not None, results are
+            filtered to directories that contain this path.
+
+        prefix:
+            String containing an absolute path or a prefix thereof. If not
+            None, results are filtered to directories starting with this
+            string.
 
     Returns:
         List of objects where each object corresponds to a registered directory
         and contains the `path` to the directory, the `user` who registered it,
         the Unix epoch `time` of the registration, and the `names` of the
         metadata files to be indexed.
     """
-    if user == True:
-        import getpass
-        user = getpass.getuser()
+    query = []
+    if not user is None and user != False:
+        if user == True:
+            import getpass
+            user = getpass.getuser()
+        query.append("user=" + user)
+    if not contains is None:
+        query.append("contains_path=" + urllib.parse.quote_plus(contains))
+    if not prefix is None:
+        query.append("path_prefix=" + urllib.parse.quote_plus(prefix))
 
     url += "/registered"
-    if not user is None and user != False:
-        url += "?user=" + user
+    if len(query) > 0:
+        url += "?" + "&".join(query)
 
     res = requests.get(url)
     if res.status_code >= 300:

src/sewerrat/start_sewerrat.py

@@ -9,7 +9,7 @@
 test_api_port = None
 
 
-def start_sewerrat(db: Optional[str] = None, port: Optional[int] = None, wait: float = 1, version: str = "1.0.9", overwrite: bool = False) -> Tuple[bool, int]:
+def start_sewerrat(db: Optional[str] = None, port: Optional[int] = None, wait: float = 1, version: str = "1.0.13", overwrite: bool = False) -> Tuple[bool, int]:
     """
     Start a test SewerRat service.
 

tests/test_registered_directories.py

@@ -23,12 +23,31 @@ def test_list_registered_directories():
                 assert x["names"] == [ "metadata.json" ]
         assert found
 
+        # Filter by user.
         filtered = sewerrat.list_registered_directories(url, user=True)
         assert regged == filtered
 
         filtered = sewerrat.list_registered_directories(url, user=regged[0]["user"] + "_asdasdasd")
         assert len(filtered) == 0
 
+        # Filter by contains.
+        filtered = sewerrat.list_registered_directories(url, contains=os.path.join(mydir, "metadata.json"))
+        assert regged == filtered
+
+        filtered = sewerrat.list_registered_directories(url, contains=os.path.join(mydir + "-asdasd"))
+        assert len(filtered) == 0
+
+        # Filter by prefix.
+        filtered = sewerrat.list_registered_directories(url, prefix=os.path.dirname(mydir))
+        assert regged == filtered
+
+        filtered  = sewerrat.list_registered_directories(url, prefix=os.path.dirname(mydir) + "-asdasdad")
+        assert len(filtered) == 0
+
+        # Multiple filters work.
+        filtered  = sewerrat.list_registered_directories(url, prefix=os.path.dirname(mydir), user=True, contains=os.path.join(mydir, "metadata.json"))
+        assert regged == filtered
+
     finally:
         sewerrat.deregister(mydir, url=url)