add env support in pipelines

rdf-connect · Feb 6, 2024 · c37777e · c37777e
1 parent 0e4628c
commit c37777e
Show file tree

Hide file tree

Showing 5 changed files with 114 additions and 3 deletions.
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@ajuvercr/js-runner",
-  "version": "0.1.13",
+  "version": "0.1.14",
   "type": "module",
   "description": "",
   "main": "./dist/index.js",

diff --git a/src/index.ts b/src/index.ts
@@ -56,6 +56,7 @@ export async function extractProcessors(
         x.object.equals(JsOntology.JsProcess),
     )
     .map((x) => x.subject);
+  console.log("Finding", JsOntology.JsChannel.value)
   const processorLens = config.lenses[JsOntology.JsProcess.value];
   const processors = subjects.map((id) => processorLens.execute({ id, quads }));
   return { processors, quads, shapes: config };
@@ -148,3 +149,4 @@ export async function jsRunner() {
     }
   }
 }
+
diff --git a/src/shacl.ts b/src/shacl.ts
@@ -10,6 +10,7 @@ import {
   subjects,
   unique,
 } from "rdf-lens";
+import { CONN2, literal } from "./util";
 
 export const RDFS = createTermNamespace(
   "http://www.w3.org/2000/01/rdf-schema#",
@@ -212,6 +213,7 @@ function optionalField<T extends string, O = string>(
       return out;
     });
 }
+
 function dataTypeToExtract(dataType: Term, t: Term): any {
   if (dataType.equals(XSD.terms.integer)) return +t.value;
   if (dataType.equals(XSD.terms.custom("float"))) return +t.value;
@@ -224,6 +226,42 @@ function dataTypeToExtract(dataType: Term, t: Term): any {
   return t;
 }
 
+function envLens(dataType: Term): BasicLens<Cont, any> {
+  const checkType = pred(RDF.terms.type)
+    .thenSome(
+      new BasicLens(({ id }) => {
+        if (!id.equals(CONN2.terms.EnvVariable)) {
+          throw "expected type " + CONN2.EnvVariable;
+        }
+        return { checked: true };
+      }),
+    )
+    .expectOne();
+
+  const envName = pred(CONN2.terms.envKey)
+    .one()
+    .map(({ id }) => ({
+      key: id.value,
+    }));
+
+  const defaultValue = pred(CONN2.terms.envDefault)
+    .one(undefined)
+    .map((found) => ({
+      defaultValue: found?.id.value,
+    }));
+
+  return checkType
+    .and(envName, defaultValue)
+    .map(([_, { key }, { defaultValue }]) => {
+      const value = process.env[key] || defaultValue;
+      if (value) {
+        return dataTypeToExtract(dataType, literal(value));
+      } else {
+        throw "Nothing set for ENV " + key + ". No default was set either!";
+      }
+    });
+}
+
 type Cache = {
   [clazz: string]: BasicLens<Cont, any>;
 };
@@ -251,7 +289,9 @@ function extractProperty(
     pred(SHACL.datatype)
       .one()
       .map(({ id }) => ({
-        extract: empty<Cont>().map((item) => dataTypeToExtract(id, item.id)),
+        extract: envLens(id).or(
+          empty<Cont>().map((item) => dataTypeToExtract(id, item.id)),
+        ),
       }));
 
   const clazzLens: BasicLens<Cont, { extract: ShapeField["extract"] }> = field(

diff --git a/src/util.ts b/src/util.ts
@@ -30,8 +30,11 @@ export const CONN2 = createUriAndTermNamespace(
   "url",
   "procFile",
   "path",
+  "EnvVariable",
+  "envKey",
+  "envDefault",
 );
-const { namedNode } = DataFactory;
+export const { namedNode, literal } = DataFactory;
 
 export type Keyed<T> = { [Key in keyof T]: Term | undefined };
 

diff --git a/test/processors.test.ts b/test/processors.test.ts
@@ -95,6 +95,72 @@ describe("test existing processors", () => {
     expect(writer.ty).toBeDefined();
   });
 
+  describe("send.ttl from env", async () => {
+    const value = `${prefixes}
+<> owl:imports <./ontology.ttl>, <./processor/send.ttl> .
+
+[ ] a :Channel;
+  :reader <jr>;
+  :writer <jw>.
+<jr> a js:JsReaderChannel.
+<jw> a js:JsWriterChannel.
+[ ] a js:Send;
+  js:msg [
+    a :EnvVariable;
+    :envDefault "FromEnv";
+    :envKey "msg"
+  ];
+  js:sendWriter <jw>.
+`;
+    const baseIRI = process.cwd() + "/config.ttl";
+    console.log(baseIRI);
+
+    const source: Source = {
+      value,
+      baseIRI,
+      type: "memory",
+    };
+
+    const {
+      processors,
+      quads,
+      shapes: config,
+    } = await extractProcessors(source);
+
+    test("Env default value", () => {
+      const proc = processors.find((x) => x.ty.value === JS + "Send");
+      expect(proc).toBeDefined();
+
+      const argss = extractSteps(proc!, quads, config);
+      expect(argss.length).toBe(1);
+      expect(argss[0].length).toBe(2);
+
+      const [[msg, writer]] = argss;
+      expect(msg).toBe("FromEnv");
+      expect(writer).toBeInstanceOf(Object);
+      expect(writer.channel).toBeDefined();
+      expect(writer.channel.id).toBeDefined();
+      expect(writer.ty).toBeDefined();
+    });
+
+    test("Env value", () => {
+      process.env["msg"] = "FROM ENV";
+      const proc = processors.find((x) => x.ty.value === JS + "Send");
+      expect(proc).toBeDefined();
+
+      const argss = extractSteps(proc!, quads, config);
+      expect(argss.length).toBe(1);
+      expect(argss[0].length).toBe(2);
+
+      const [[msg, writer]] = argss;
+      expect(msg).toBe("FROM ENV");
+      expect(writer).toBeInstanceOf(Object);
+      expect(writer.channel).toBeDefined();
+      expect(writer.channel.id).toBeDefined();
+      expect(writer.ty).toBeDefined();
+    });
+  });
+
   test("echo.ttl", async () => {
     const value = `${prefixes}
 <> owl:imports <./ontology.ttl>, <./processor/echo.ttl> .