Merge pull request #87 from edanalytics/feature/namespace_overrides

adding `namespace_overrides` feature

Author: tomreitz

CHANGELOG.md

@@ -1,3 +1,9 @@
+### v0.1.11
+<details>
+<summary>Released 2026-02-17</summary>
+* feature: `namespace_overrides` to allow sending resources to different namespaces in one `lightbeam send`
+</details>
+
 ### v0.1.10
 <details>
 <summary>Released 2025-11-21</summary>

README.md

@@ -41,6 +41,9 @@ An example YAML configuration is below, followed by documentation of each option
 state_dir: ~/.lightbeam/
 data_dir: ./
 namespace: ed-fi
+namespace_overrides:
+  tpdm:
+    - candidates
 edfi_api:
   base_url: https://api.schooldistrict.org/v5.3/api
   oauth_url: https://api.schooldistrict.org/v5.3/api/oauth/token 
@@ -84,6 +87,7 @@ show_stacktrace: True
 * (optional) `state_dir` is where [state](#state) is stored. The default is `~/.lightbeam/` on *nix systems, `C:/Users/USER/.lightbeam/` on Windows systems.
 * (optional) Specify the `data_dir` which contains JSONL files to send to Ed-Fi. The default is `./`. The tool will look for files like `{Resource}.jsonl` or `{Descriptor}.jsonl` in this location, as well as directory-based files like `{Resource}/*.jsonl` or `{Descriptor}/*.jsonl`. Files with `.ndjson` or simply `.json` extensions will also be processed. (More info at the [`ndjson` standard page](http://dataprotocols.org/ndjson/).)
 * (optional) Specify the `namespace` to use when accessing the Ed-Fi API. The default is `ed-fi` but others include `tpdm` or custom values. To send data to multiple namespaces, you must use a YAML configuration file and `lightbeam send` for each.
+* (optional) Specify `namespace_overrides`: a structure where keys are alternate namespaces (beside the above `namespace`) and values are lists of endpoint names that correspond to that namespace. This enables lightbeam to map data files for different endpoints to different namespaces, so you can (for example) transmit `candidates.jsonl` to the `tpdm` namespace and `staffs.jsonl` to the `ed-fi` namespace in a single `lightbeam send`.
 * Specify the details of the `edfi_api` to which to connect including
   * (optional) The `base_url` which serves a JSON object specifying the paths to data endpoints, Swagger, and dependencies. The default is `https://localhost/api` (the address of an Ed-Fi API [running locally in Docker](https://docs.ed-fi.org/reference/docker/)), but the location varies depending on how Ed-Fi is deployed.
   * Most Ed-Fi APIs publish other endpoints they provide in a JSON document at the base URL - `lighbteam` attempts to discover these. If, however, your API does not publish these URLs at the base, or you want to manually override any of them for any reason, you can specify the following:

example/lightbeam.yml

@@ -1,6 +1,9 @@
 # state_dir: ~/.lightbeam/
 data_dir: ./
 namespace: ed-fi
+namespace_overrides:
+  tpdm:
+    - candidates
 edfi_api:
   # base_url: https://api.ed-fi.org/v5.3/api
   base_url: https://localhost/api

lightbeam/VERSION.txt

@@ -1 +1 @@
-0.1.10
+0.1.11

lightbeam/api.py

@@ -179,9 +179,15 @@ def get_sorted_endpoints(self):
         data = sorted(data, key=lambda x: x['order'])
 
         ordered_endpoints = []
+        possible_namespaces = [self.lightbeam.config["namespace"]]
+        if "namespace_overrides" in self.lightbeam.config.keys():
+            for namespace in self.lightbeam.config["namespace_overrides"].keys():
+                if namespace == "__line__": continue # (remove YAML parsing artifact)
+                possible_namespaces.append(namespace)
         for e in data:
-            if e["resource"].startswith("/" + self.lightbeam.config["namespace"] + "/"):
-                ordered_endpoints.append(e["resource"].replace('/' + self.lightbeam.config["namespace"] + '/', ""))
+            for namespace in possible_namespaces:
+                if e["resource"].startswith(f"/{namespace}/"):
+                    ordered_endpoints.append(e["resource"].replace(f"/{namespace}/", ""))
         return ordered_endpoints
 
     # Loads the Swagger JSON from the Ed-Fi API
@@ -346,7 +352,7 @@ async def load_descriptors_values(self):
     def get_params_for_endpoint(self, endpoint, type='required'):
         if "Descriptor" in endpoint: swagger = self.descriptors_swagger
         else: swagger = self.resources_swagger
-        definition = util.get_swagger_ref_for_endpoint(self.lightbeam.config["namespace"], swagger, endpoint)
+        definition = util.get_swagger_ref_for_endpoint(self.lightbeam.get_namespace_for_endpoint(endpoint), swagger, endpoint)
         if type=='required':
             return self.get_required_params_from_swagger(swagger, definition)
         elif type=='all':

lightbeam/count.py

@@ -55,7 +55,7 @@ async def get_record_count(self, endpoint, params={}):
             # since `count` should finish _very_ quickly - way before expiry
             params.update({ "limit": "0", "totalCount": "true" })
             async with self.lightbeam.api.client.get(
-                util.url_join(self.lightbeam.api.config["data_url"], self.lightbeam.config["namespace"], endpoint),
+                util.url_join(self.lightbeam.api.config["data_url"], self.lightbeam.get_namespace_for_endpoint(endpoint), endpoint),
                 params=params,
                 ssl=self.lightbeam.config["connection"]["verify_ssl"],
                 headers=self.lightbeam.api.headers

lightbeam/delete.py

@@ -141,7 +141,7 @@ async def do_delete(self, endpoint, file_name, params, line, data_hash=None):
 
                     # we have to get the `id` for a particular resource by first searching for its natural keys
                     async with self.lightbeam.api.client.get(
-                        util.url_join(self.lightbeam.api.config["data_url"], self.lightbeam.config["namespace"], endpoint),
+                        util.url_join(self.lightbeam.api.config["data_url"], self.lightbeam.get_namespace_for_endpoint(endpoint), endpoint),
                         params=params,
                         ssl=self.lightbeam.config["connection"]["verify_ssl"],
                         headers=self.lightbeam.api.headers
@@ -192,7 +192,7 @@ async def do_delete_id(self, endpoint, id, file_name=None, line=None, data_hash=
         while True: # this is not great practice, but an effective way (along with the `break` below) to achieve a do:while loop
             try:
                 async with self.lightbeam.api.client.delete(
-                    util.url_join(self.lightbeam.api.config["data_url"], self.lightbeam.config["namespace"], endpoint, id),
+                    util.url_join(self.lightbeam.api.config["data_url"], self.lightbeam.get_namespace_for_endpoint(endpoint), endpoint, id),
                     ssl=self.lightbeam.config["connection"]["verify_ssl"],
                     headers=self.lightbeam.api.headers
                     ) as delete_response:

lightbeam/fetch.py

@@ -31,7 +31,7 @@ async def get_records(self, do_write=True, log_status_counts=True):
             if self.lightbeam.query != '':
                 self.lightbeam.api.load_swagger_docs()
                 swagger = self.lightbeam.api.resources_swagger
-                namespace = self.lightbeam.config["namespace"]
+                namespace = self.lightbeam.get_namespace_for_endpoint(endpoint)
                 supported_params = swagger.get("paths", {}).get(f"/{namespace}/{endpoint}", {}).get("get", {}).get("parameters", [])
                 supported_param_names = [ x["name"] for x in supported_params if "name" in x.keys() and "in" in x.keys() and x["in"]=="query" ]
                 if not set(params.keys()).issubset(set(supported_param_names)):
@@ -76,7 +76,7 @@ async def get_endpoint_records(self, endpoint, limit, offset, file_handle=None):
 
                 # send GET request
                 async with self.lightbeam.api.client.get(
-                    util.url_join(self.lightbeam.api.config["data_url"], self.lightbeam.config["namespace"], endpoint),
+                    util.url_join(self.lightbeam.api.config["data_url"], self.lightbeam.get_namespace_for_endpoint(endpoint), endpoint),
                     params=urlencode(params),
                     ssl=self.lightbeam.config["connection"]["verify_ssl"],
                     headers=self.lightbeam.api.headers

lightbeam/lightbeam.py

@@ -125,13 +125,16 @@ def __init__(self, config_file, logger=None, selector="*", exclude="", keep_keys
             os.mkdir(self.config["state_dir"])
 
         # Initialize a dictionary for tracking run metadata (for structured output)
+        namespace_overrides = self.config["namespace_overrides"] if "namespace_overrides" in self.config.keys() else None
+        namespace_overrides.pop("__line__") # (remove YAML parsing artifact)
         self.metadata = {
             "started_at": self.start_timestamp.isoformat(timespec='microseconds'),
             "working_dir": os.getcwd(),
             "config_file": self.config_file,
             "data_dir": self.config["data_dir"],
             "api_url": self.config["edfi_api"]["base_url"],
             "namespace": self.config["namespace"],
+            "namespace_overrides": namespace_overrides,
             "resources": {}
         }
 
@@ -250,6 +253,16 @@ def confirm_delete(self, endpoints):
     def confirm_truncate(self, endpoints):
         self._confirm_delete_op(endpoints, "TRUNCATE ALL DATA")
 
+    def get_namespace_for_endpoint(self, endpoint):
+        if "namespace_overrides" in self.config.keys():
+            for namespace in self.config["namespace_overrides"].keys():
+                if (
+                    isinstance(self.config["namespace_overrides"][namespace], list)
+                    and endpoint in self.config["namespace_overrides"][namespace]
+                ):
+                    return namespace
+        return self.config["namespace"]
+
     ################### Data discovery and loading methods ####################
 
     # For the specified endpoint, returns a list of all files in config.data_dir which end in .jsonl

lightbeam/send.py

@@ -142,7 +142,7 @@ async def do_post(self, endpoint, file_name, data, line_number, data_hash):
         while True: # this is not great practice, but an effective way (along with the `break` below) to achieve a do:while loop
             try:
                 async with self.lightbeam.api.client.post(
-                    util.url_join(self.lightbeam.api.config["data_url"], self.lightbeam.config["namespace"], endpoint),
+                    util.url_join(self.lightbeam.api.config["data_url"], self.lightbeam.get_namespace_for_endpoint(endpoint), endpoint),
                     data=data,
                     ssl=self.lightbeam.config["connection"]["verify_ssl"],
                     headers=self.lightbeam.api.headers