add batch size to nabu

.env.example

@@ -17,6 +17,7 @@ GLEANER_THREADS=5
 NABU_PROFILING=false
 # log levels should be in all caps
 NABU_LOG_LEVEL=INFO
+NABU_BATCH_SIZE=100
 
 
 # Minio

userCode/lib/containers.py

@@ -4,7 +4,7 @@
 import os
 from pathlib import Path
 from userCode.lib.types import cli_modes
-from userCode.lib.env import GLEANER_IMAGE, NABU_IMAGE, NABU_PROFILING
+from userCode.lib.env import GLEANER_IMAGE, NABU_BATCH_SIZE, NABU_IMAGE, NABU_PROFILING
 from userCode.lib.utils import run_scheduler_docker_image
 
 
@@ -18,9 +18,9 @@ def __init__(
         self.name = "gleaner"
         self.source = source
 
-        assert Path(
-            "/tmp/geoconnex/"
-        ).exists(), "the /tmp/geoconnex directory does not exist. This must exist for us to share configs with the docker socket on the host"
+        assert Path("/tmp/geoconnex/").exists(), (
+            "the /tmp/geoconnex directory does not exist. This must exist for us to share configs with the docker socket on the host"
+        )
 
     def run(self, args: list[str]):
         run_scheduler_docker_image(
@@ -43,6 +43,8 @@ def run(self, args: list[str]):
         if NABU_PROFILING:
             args.append("--trace")
 
+        args.append(f"--upsert-batch-size={NABU_BATCH_SIZE}")
+
         nabu_log_level = os.environ.get("NABU_LOG_LEVEL")
         if nabu_log_level:
             args.append(f"--log-level={nabu_log_level}")

userCode/lib/env.py

@@ -104,9 +104,10 @@ def strict_env(key: str):
 
 DAGSTER_YAML_CONFIG: str = os.path.join(userCodeRoot, "dagster.yaml")
 
-assert Path(
-    DAGSTER_YAML_CONFIG
-).exists(), f"the dagster.yaml file does not exist at {DAGSTER_YAML_CONFIG}"
+assert Path(DAGSTER_YAML_CONFIG).exists(), (
+    f"the dagster.yaml file does not exist at {DAGSTER_YAML_CONFIG}"
+)
 
 
 NABU_PROFILING = strict_env("NABU_PROFILING")
+NABU_BATCH_SIZE = strict_env_int("NABU_BATCH_SIZE")