Skip to content

Commit

Permalink
refactor: add doc and warning on not found manifest with loadManifest (…
Browse files Browse the repository at this point in the history 
…#155)

close #154
  • Loading branch information
@mgabelle
mgabelle authored Oct 25, 2024
1 parent 4d0b183 commit bf2d485
Showing 1 changed file with 79 additions and 17 deletions.
96 changes: 79 additions & 17 deletions src/main/java/io/kestra/plugin/dbt/cli/DbtCLI.java
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
package io.kestra.plugin.dbt.cli;

import com.fasterxml.jackson.databind.ObjectMapper;
import io.kestra.core.exceptions.IllegalVariableEvaluationException;
import io.kestra.core.exceptions.ResourceExpiredException;
import io.kestra.core.models.annotations.Example;
import io.kestra.core.models.annotations.Plugin;
import io.kestra.core.models.annotations.PluginProperty;
Expand All @@ -11,6 +13,7 @@
import io.kestra.core.runners.RunContext;
import io.kestra.core.serializers.JacksonMapper;
import io.kestra.core.storages.kv.KVStore;
import io.kestra.core.storages.kv.KVValue;
import io.kestra.core.storages.kv.KVValueAndMetadata;
import io.kestra.plugin.dbt.ResultParser;
import io.kestra.plugin.scripts.exec.AbstractExecScript;
Expand All @@ -25,15 +28,19 @@
import org.apache.commons.io.FileUtils;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;

import jakarta.validation.constraints.NotEmpty;
import jakarta.validation.constraints.NotNull;
import org.apache.commons.lang3.StringUtils;

@SuperBuilder
@ToString
Expand Down Expand Up @@ -148,12 +155,6 @@
- dbt deps --project-dir dbt --target prod
- dbt build --project-dir dbt --target prod
projectDir: dbt
loadManifest:
key: manifest.json
namespace: company.team
storeManifest:
key: manifest.json
namespace: company.team
profiles: |
my_dbt_project:
outputs:
Expand All @@ -174,6 +175,56 @@
threads: 16
timeout_seconds: 300
target: dev"""
),
@Example(
title = "Clone a [Git repository](https://github.com/kestra-io/dbt-example) and build dbt models in defer mode. The loadManifest property will fetch an existing manifest.json and use it for the defer build.",
full = true,
code = """
id: dbt_duckdb
namespace: company.team
tasks:
- id: dbt
type: io.kestra.plugin.core.flow.WorkingDirectory
tasks:
- id: clone_repository
type: io.kestra.plugin.git.Clone
url: https://github.com/kestra-io/dbt-example
branch: main
- id: dbt_build
type: io.kestra.plugin.dbt.cli.DbtCLI
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
delete: true
containerImage: ghcr.io/kestra-io/dbt-duckdb:latest
loadManifest:
key: manifest.json
namespace: company.team
storeManifest:
key: manifest.json
namespace: company.team
commands:
- dbt build --defer --state ./target --target prod
profiles: |
my_dbt_project:
outputs:
dev:
type: duckdb
path: ":memory:"
fixed_retries: 1
threads: 16
timeout_seconds: 300
prod:
type: duckdb
path: dbt2.duckdb
extensions:
- parquet
fixed_retries: 1
threads: 16
timeout_seconds: 300
target: dev
"""
)
}
)
Expand Down Expand Up @@ -233,7 +284,10 @@ public class DbtCLI extends AbstractExecScript {

@Schema(
title = "Load manifest.",
description = "Use this field to retrieve an existing manifest.json in the KV Store and put it in the inputFiles."
description = """
Use this field to retrieve an existing manifest.json in the KV Store and put it in the inputFiles.
The manifest.json will be put under ./target/manifest.json or under ./projectDir/target/manifest.json if you specify a projectDir.
"""
)
protected KvStoreManifest loadManifest;

Expand Down Expand Up @@ -277,15 +331,7 @@ public void accept(String line, Boolean isStdErr) {
//Load manifest from KV store
if(this.getLoadManifest() != null) {
KVStore loadManifestKvStore = runContext.namespaceKv(this.getLoadManifest().getNamespace().as(runContext, String.class));
var manifestFile = new File(projectWorkingDirectory.toString(), "target/manifest.json");
Object manifestValue = loadManifestKvStore.getValue(this.getLoadManifest().getKey().as(runContext, String.class)).get().value();

FileUtils.writeStringToFile(
manifestFile,
JacksonMapper.ofJson()
.writeValueAsString(manifestValue),
StandardCharsets.UTF_8
);
fetchAndStoreManifestIfExists(runContext, loadManifestKvStore, projectWorkingDirectory);
}

//Create profiles.yml
Expand Down Expand Up @@ -331,7 +377,7 @@ public void accept(String line, Boolean isStdErr) {

File manifestFile = projectWorkingDirectory.resolve("target/manifest.json").toFile();
if (manifestFile.exists()) {
if(this.storeManifest != null) {
if(this.getStoreManifest() != null) {
final String key = this.getStoreManifest().getKey().as(runContext, String.class);
storeManifestKvStore.put(key, new KVValueAndMetadata(null, JacksonMapper.toObject(Files.readString(manifestFile.toPath()))));
}
Expand All @@ -344,6 +390,22 @@ public void accept(String line, Boolean isStdErr) {
return run;
}

private void fetchAndStoreManifestIfExists(RunContext runContext, KVStore loadManifestKvStore, Path projectWorkingDirectory) throws IOException, ResourceExpiredException, IllegalVariableEvaluationException {
Optional<KVValue> manifestValue = loadManifestKvStore.getValue(this.getLoadManifest().getKey().as(runContext, String.class));

if(manifestValue.isEmpty() || manifestValue.get().value() == null || StringUtils.isBlank(manifestValue.get().value().toString())) {
runContext.logger().warn("Property `loadManifest` has been used but no manifest has been found in the KV Store.");
return;
}
var manifestFile = new File(projectWorkingDirectory.toString(), "target/manifest.json");
FileUtils.writeStringToFile(
manifestFile,
JacksonMapper.ofJson()
.writeValueAsString(manifestValue.get().value()),
StandardCharsets.UTF_8
);
}

@Builder
@Getter
public static class KvStoreManifest {
Expand Down

0 comments on commit bf2d485

Please sign in to comment.