Merge pull request #9 from netwerk-digitaal-erfgoed/issue-8

closes issue #8

src/lib/File.class.ts

@@ -1,10 +1,11 @@
-import { type WriteStream, createWriteStream, existsSync, statSync } from 'fs'
+import { type WriteStream, createWriteStream, existsSync, statSync, mkdirSync } from 'fs'
 import { isFile, isFilePathString } from '../utils/guards.js'
+import { dirname } from 'path'
 
 export default class File {
  public static $id = 'File'
  private readonly $isValid?: boolean
- public constructor(private $path: string) {}
+ public constructor(private $path: string, private readonly skipExistsCheck: boolean = false) {}
 
  public validate(): File {
  if (this.$isValid !== undefined) return this
@@ -13,7 +14,7 @@ export default class File {
  throw new Error(`The filename \`${wrongFilePath}\` should start with \`file://\``)
  }
  this.$path = this.$path.replace(/^file:\/\//, '')
- if (!existsSync(this.$path) || !statSync(this.$path).isFile()) {
+ if (!this.skipExistsCheck && (!existsSync(this.$path) || !statSync(this.$path).isFile())) {
  throw new Error(`File not found: \`${this.$path}\``)
  }
  return this
@@ -23,11 +24,14 @@ export default class File {
  return this.$path
  }
 
- public getStream(): WriteStream {
+ public getStream(append: boolean = false): WriteStream {
  if (existsSync(this.$path)) {
  // throw new Error(`File already exists: \`${this.$path}\``)
  }
- return createWriteStream(this.$path)
+ if (!existsSync(dirname(this.$path))) {
+ mkdirSync(dirname(this.$path), { recursive: true})
+ }
+ return createWriteStream(this.$path, append ? {flags: 'a'} : {})
  }
 
  public toString(): string {

src/lib/LDWorkbenchConfiguration.d.ts

@@ -21,7 +21,7 @@ export interface LDWorkbenchConfiguration {
  /**
  * The file where the final result of your pipeline is saved.
  */
- destination: string;
+ destination?: string;
  /**
  * This is where you define the individual iterator/generator for each step.
  *

src/lib/Pipeline.class.ts

@@ -4,22 +4,33 @@ import type { LDWorkbenchConfiguration } from "./LDWorkbenchConfiguration.js";
 import chalk from "chalk";
 import Stage from "./Stage.class.js";
 import duration from "../utils/duration.js";
+import File from './File.class.js'
 import path from "node:path";
 import * as fs from "node:fs";
+import { isFilePathString } from '../utils/guards.js';
 
 class Pipeline {
  public readonly stages = new Map<string, Stage>();
  public dataDir: string;
  private $isValidated: boolean = false;
  private stageNames: string[] = [];
  private now = new Date();
+ private readonly destination: File
 
  public constructor(
  private readonly $configuration: LDWorkbenchConfiguration
  ) {
  // create data folder:
  this.dataDir = path.join("data", kebabcase(this.$configuration.name));
  fs.mkdirSync(this.dataDir, { recursive: true });
+ const destinationFile = this.configuration.destination ?? `file://${path.join(this.dataDir, 'statements.nt')}`
+ if (!isFilePathString(destinationFile)) {
+ throw new Error('We currently only allow publishing data to local files.')
+ }
+ if(!destinationFile.endsWith('.nt')) {
+ throw new Error('We currently only writing results in N-Triples format,\nmake sure your destination filename ends with \'.nt\'.')
+ }
+ this.destination = new File(destinationFile, true)
  }
 
  private error(e: Error, stage?: string): void {
@@ -142,7 +153,7 @@ class Pipeline {
  if (this.stageNames.length !== 0) {
  this.runRecursive();
  } else {
- this.concat()
+ this.writeResult()
  console.info(
  chalk.green(
  `✔ your pipeline "${chalk.bold(
@@ -159,10 +170,14 @@ class Pipeline {
  }
  }
 
- private concat(): void {
+ private writeResult(): void {
  const spinner = ora("Combining statements from all stages:").start();
- const destinationPath = path.join(this.dataDir, 'statements.nt')
- const destinationStream = fs.createWriteStream(destinationPath, {flags:'a'})
+
+ const destinationPathNew = this.configuration.destination
+ if (!isFilePathString(destinationPathNew)) {
+ throw new Error('We currently only allow publishing data to local files.')
+ }
+ const destinationStream = this.destination.getStream()
  const stageNames = Array.from(this.stages.keys())
  for (const stageName of stageNames) {
  spinner.suffixText = chalk.bold(stageName)
@@ -171,7 +186,7 @@ class Pipeline {
  destinationStream.write(buffer)
  })
  }
- spinner.suffixText = chalk.bold(destinationPath)
+ spinner.suffixText = chalk.bold(this.destination.toString())
  spinner.succeed()
  }
 

static/example/config.yml

@@ -1,14 +1,13 @@
 # Metadata for your pipeline:
 name: Example Pipeline
-destination: file://data/example/example.ttl
 description: >
  This is an example pipeline. It uses files that are available in this repository 
  and SPARQL endpoints that should work.
 
 # The individual stages for your pipeline
 stages:
  - name: "Stage 1"
- iterator: 
+ iterator:
  query: file://static/example/iterator-stage-1.rq
  endpoint: https://api.triplydb.com/datasets/Triply/iris/services/demo-service/sparql
  generator: 

static/ld-workbench.schema.json

@@ -77,5 +77,5 @@
  }
  }
  },
- "required": ["name", "destination", "stages"]
+ "required": ["name", "stages"]
 }