Make key name optional when a regex is used in the config

Ouranosinc · Nov 15, 2023 · 2dc57c8 · 2dc57c8
1 parent 39a805b
commit 2dc57c8
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 21 deletions.
diff --git a/cowbird/config.py b/cowbird/config.py
@@ -4,7 +4,7 @@
 from typing import Any, Dict, List, Literal, Tuple, Union, cast, overload
 
 import yaml
-from schema import And, Optional, Regex, Schema
+from schema import And, Optional, Or, Regex, Schema
 
 from cowbird.typedefs import (
     ConfigDict,
@@ -191,7 +191,7 @@ def validate_sync_perm_config_schema(sync_cfg: SyncPointConfig) -> None:
             "services": {
                 str: {  # Service type, must correspond to an actual Magpie service type
                     str: [  # Resource key, used to identify the resource here and in the permissions_mapping
-                        {"name": str, "type": str, Optional("field"): str, Optional("regex"): str}
+                        {Or("name", "regex", only_one=True): str, "type": str, Optional("field"): str}
                     ]
                 }
             },
@@ -213,21 +213,23 @@ def validate_and_get_resource_info(res_key: str, segments: List[ConfigSegment])
     named_tokens = set()
     has_multi_token = False
     for seg in segments:
-        if seg["name"] == MULTI_TOKEN:
-            if has_multi_token:
-                raise ConfigErrorInvalidTokens(f"Invalid config value for resource key {res_key}. Only one "
-                                               f"`{MULTI_TOKEN}` token is permitted per resource.")
-            has_multi_token = True
-        else:
-            matched_groups = re.match(NAMED_TOKEN_REGEX, seg["name"])
-            if matched_groups:
-                # Save the first group as a named token, since there's only 1 matching group in the regex.
-                if matched_groups.groups()[0] in named_tokens:
-                    raise ConfigErrorInvalidTokens(f"Invalid config value for resource key {res_key}. Named token "
-                                                   f"{matched_groups.groups()[0]} was found in multiple segments of "
-                                                   "the resource path. Each named token should only be used once in a "
-                                                   "resource path.")
-                named_tokens.add(matched_groups.groups()[0])
+        if "name" in seg:
+            if seg["name"] == MULTI_TOKEN:
+                if has_multi_token:
+                    raise ConfigErrorInvalidTokens(f"Invalid config value for resource key {res_key}. Only one "
+                                                   f"`{MULTI_TOKEN}` token is permitted per resource.")
+                has_multi_token = True
+            else:
+                matched_groups = re.match(NAMED_TOKEN_REGEX, seg["name"])
+                if matched_groups:
+                    # Save the first group as a named token, since there's only 1 matching group in the regex.
+                    if matched_groups.groups()[0] in named_tokens:
+                        raise ConfigErrorInvalidTokens(
+                            f"Invalid config value for resource key {res_key}. Named token "
+                            f"{matched_groups.groups()[0]} was found in multiple segments of "
+                            "the resource path. Each named token should only be used once in a "
+                            "resource path.")
+                    named_tokens.add(matched_groups.groups()[0])
 
     return {"has_multi_token": has_multi_token, "named_tokens": named_tokens}
 

diff --git a/cowbird/permissions_synchronizer.py b/cowbird/permissions_synchronizer.py
@@ -217,11 +217,15 @@ def _generate_nametype_path_from_segments(res_segments: List[ConfigSegment],
         :param src_resource_tree: Resource tree associated with the permission to synchronize
         """
         resource_nametype_path = ""
-        for res_seg, res in zip(res_segments, src_resource_tree):
-            key = res_seg.get("field")
-            if key is None:
-                key = "resource_name"
+        res_segments_len = len(res_segments)
+        for index, res in enumerate(src_resource_tree):
             res_type = res["resource_type"]
+            if index < res_segments_len:
+                key = res_segments[index].get("field")
+                if key is None:
+                    key = "resource_name"
+            else:
+                key = "resource_name"
             resource_nametype_path += f"/{res[key]}{RES_NAMETYPE_SEPARATOR}{res_type}"
 
         return resource_nametype_path