Skip to content

Commit 23f11c5

Browse files
aeSouidaeSouid
committed
Migrate to the kestra http client instead of micronaut
1 parent 318d5e7 commit 23f11c5

File tree

3 files changed

+78
-131
lines changed

3 files changed

+78
-131
lines changed

‎src/main/java/io/kestra/plugin/dbt/cloud/AbstractDbtCloud.java

Lines changed: 45 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -1,98 +1,81 @@
11
package io.kestra.plugin.dbt.cloud;
22

3+
import com.fasterxml.jackson.databind.DeserializationFeature;
4+
import com.fasterxml.jackson.databind.ObjectMapper;
5+
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
36
import io.kestra.core.exceptions.IllegalVariableEvaluationException;
47
import io.kestra.core.models.property.Property;
58
import io.kestra.core.models.tasks.Task;
6-
import io.kestra.core.runners.DefaultRunContext;
79
import io.kestra.core.runners.RunContext;
8-
import io.micronaut.core.type.Argument;
9-
import io.micronaut.http.HttpResponse;
10-
import io.micronaut.http.MediaType;
11-
import io.micronaut.http.MutableHttpRequest;
12-
import io.micronaut.http.client.DefaultHttpClientConfiguration;
13-
import io.micronaut.http.client.HttpClient;
14-
import io.micronaut.http.client.exceptions.HttpClientResponseException;
15-
import io.micronaut.http.client.netty.DefaultHttpClient;
16-
import io.micronaut.http.client.netty.NettyHttpClientFactory;
17-
import io.micronaut.http.codec.MediaTypeCodecRegistry;
10+
import io.kestra.core.http.HttpRequest;
11+
import io.kestra.core.http.HttpResponse;
12+
import io.kestra.core.http.client.HttpClient;
13+
import io.kestra.core.http.client.HttpClientException;
14+
import io.kestra.core.http.client.configurations.HttpConfiguration;
15+
1816
import io.swagger.v3.oas.annotations.media.Schema;
1917
import lombok.*;
2018
import lombok.experimental.SuperBuilder;
2119

22-
import java.net.MalformedURLException;
23-
import java.net.URI;
24-
import java.net.URISyntaxException;
25-
import java.time.Duration;
20+
import java.io.IOException;
21+
2622
import jakarta.validation.constraints.NotNull;
27-
import reactor.core.publisher.Mono;
2823

2924
@SuperBuilder
3025
@ToString
3126
@EqualsAndHashCode
3227
@Getter
3328
@NoArgsConstructor
3429
public abstract class AbstractDbtCloud extends Task {
35-
@Schema(
36-
title = "Base url to select the tenant."
37-
)
30+
private static final ObjectMapper MAPPER = new ObjectMapper()
31+
.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
32+
.registerModule(new JavaTimeModule());
33+
34+
@Schema(title = "Base URL to select the tenant.")
3835
@NotNull
3936
@Builder.Default
4037
Property<String> baseUrl = Property.of("https://cloud.getdbt.com");
4138

42-
@Schema(
43-
title = "Numeric ID of the account."
44-
)
39+
@Schema(title = "Numeric ID of the account.")
4540
@NotNull
4641
Property<String> accountId;
4742

48-
@Schema(
49-
title = "API key."
50-
)
43+
@Schema(title = "API key.")
5144
@NotNull
5245
Property<String> token;
5346

54-
private static final Duration HTTP_READ_TIMEOUT = Duration.ofSeconds(60);
55-
private static final NettyHttpClientFactory FACTORY = new NettyHttpClientFactory();
47+
@Schema(title = "The HTTP client configuration.")
48+
HttpConfiguration options;
5649

57-
protected HttpClient client(RunContext runContext) throws IllegalVariableEvaluationException, MalformedURLException, URISyntaxException {
58-
MediaTypeCodecRegistry mediaTypeCodecRegistry = ((DefaultRunContext)runContext).getApplicationContext().getBean(MediaTypeCodecRegistry.class);
50+
/**
51+
* Perform an HTTP request using Kestra HttpClient.
52+
*
53+
* @param requestBuilder The prepared HTTP request builder.
54+
* @param responseType The expected response type.
55+
* @param <RES> The response class.
56+
* @return HttpResponse of type RES.
57+
*/
58+
protected <RES> HttpResponse<RES> request(RunContext runContext, HttpRequest.HttpRequestBuilder requestBuilder, Class<RES> responseType)
59+
throws HttpClientException, IllegalVariableEvaluationException {
5960

60-
var httpConfig = new DefaultHttpClientConfiguration();
61-
httpConfig.setMaxContentLength(Integer.MAX_VALUE);
62-
httpConfig.setReadTimeout(HTTP_READ_TIMEOUT);
61+
var request = requestBuilder
62+
.addHeader("Authorization", "Bearer " + runContext.render(this.token).as(String.class).orElseThrow())
63+
.addHeader("Content-Type", "application/json")
64+
.build();
6365

64-
DefaultHttpClient client = (DefaultHttpClient) FACTORY.createClient(URI.create(runContext.render(baseUrl).as(String.class).orElseThrow()).toURL(), httpConfig);
65-
client.setMediaTypeCodecRegistry(mediaTypeCodecRegistry);
66+
try (HttpClient client = new HttpClient(runContext, options)) {
67+
HttpResponse<String> response = client.request(request, String.class);
6668

67-
return client;
68-
}
69-
70-
protected <REQ, RES> HttpResponse<RES> request(RunContext runContext,
71-
MutableHttpRequest<REQ> request,
72-
Argument<RES> argument) throws HttpClientResponseException {
73-
return request(runContext, request, argument, null);
74-
}
75-
protected <REQ, RES> HttpResponse<RES> request(RunContext runContext,
76-
MutableHttpRequest<REQ> request,
77-
Argument<RES> argument,
78-
Duration timeout) throws HttpClientResponseException {
79-
try {
80-
request = request
81-
.bearerAuth(runContext.render(this.token).as(String.class).orElseThrow())
82-
.contentType(MediaType.APPLICATION_JSON);
69+
RES parsedResponse = MAPPER.readValue(response.getBody(), responseType);
70+
return HttpResponse.<RES>builder()
71+
.request(request)
72+
.body(parsedResponse)
73+
.headers(response.getHeaders())
74+
.status(response.getStatus())
75+
.build();
8376

84-
try (HttpClient client = this.client(runContext)) {
85-
Mono<HttpResponse<RES>> mono = Mono.from(client.exchange(request, argument));
86-
return timeout != null ? mono.block(timeout) : mono.block();
87-
}
88-
} catch (HttpClientResponseException e) {
89-
throw new HttpClientResponseException(
90-
"Request failed '" + e.getStatus().getCode() + "' and body '" + e.getResponse().getBody(String.class).orElse("null") + "'",
91-
e,
92-
e.getResponse()
93-
);
94-
} catch (IllegalVariableEvaluationException | MalformedURLException | URISyntaxException e) {
95-
throw new RuntimeException(e);
77+
} catch (IOException e) {
78+
throw new RuntimeException("Error executing HTTP request", e);
9679
}
9780
}
9881
}

‎src/main/java/io/kestra/plugin/dbt/cloud/CheckStatus.java

Lines changed: 16 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
package io.kestra.plugin.dbt.cloud;
22

33
import io.kestra.core.exceptions.IllegalVariableEvaluationException;
4+
import io.kestra.core.http.HttpRequest;
5+
import io.kestra.core.http.client.HttpClientException;
46
import io.kestra.core.models.annotations.Example;
57
import io.kestra.core.models.annotations.Plugin;
68
import io.kestra.core.models.property.Property;
@@ -11,10 +13,7 @@
1113
import io.kestra.plugin.dbt.cloud.models.JobStatusHumanizedEnum;
1214
import io.kestra.plugin.dbt.cloud.models.RunResponse;
1315
import io.kestra.plugin.dbt.cloud.models.Step;
14-
import io.micronaut.core.type.Argument;
15-
import io.micronaut.http.HttpMethod;
16-
import io.micronaut.http.HttpRequest;
17-
import io.micronaut.http.uri.UriTemplate;
16+
1817
import io.swagger.v3.oas.annotations.media.Schema;
1918
import jakarta.validation.constraints.NotNull;
2019
import lombok.*;
@@ -192,53 +191,23 @@ private void logSteps(Logger logger, RunResponse runResponse) {
192191
}
193192
}
194193

195-
private Optional<RunResponse> fetchRunResponse(RunContext runContext, Long id, Boolean debug) throws IllegalVariableEvaluationException {
196-
return this
197-
.request(
198-
runContext,
199-
HttpRequest
200-
.create(
201-
HttpMethod.GET,
202-
UriTemplate
203-
.of("/api/v2/accounts/{accountId}/runs/{runId}/" +
204-
"?include_related=" + URLEncoder.encode(
205-
"[\"trigger\",\"job\"," + (debug ? "\"debug_logs\"" : "") + ",\"run_steps\", \"environment\"]",
206-
StandardCharsets.UTF_8
207-
)
208-
)
209-
.expand(Map.of(
210-
"accountId", runContext.render(this.accountId).as(String.class).orElseThrow(),
211-
"runId", id
212-
))
213-
),
214-
Argument.of(RunResponse.class),
215-
runContext.render(this.maxDuration).as(Duration.class).orElseThrow()
216-
)
217-
.getBody();
194+
private Optional<RunResponse> fetchRunResponse(RunContext runContext, Long id, Boolean debug) throws IllegalVariableEvaluationException, HttpClientException {
195+
HttpRequest.HttpRequestBuilder requestBuilder = HttpRequest.builder()
196+
.uri(URI.create(runContext.render(this.baseUrl).as(String.class).orElseThrow() + "/api/v2/accounts/" + runContext.render(this.accountId).as(String.class).orElseThrow() + "/runs/" + id +
197+
"/?include_related=" + URLEncoder.encode("[\"trigger\",\"job\"," + (debug ? "\"debug_logs\"" : "") + ",\"run_steps\", \"environment\"]", StandardCharsets.UTF_8)))
198+
.method("GET");
199+
200+
return Optional.ofNullable(this.request(runContext, requestBuilder, RunResponse.class).getBody());
218201
}
219202

220-
private Path downloadArtifacts(RunContext runContext, Long runId, String path) throws IllegalVariableEvaluationException, IOException {
221-
String artifact = this
222-
.request(
223-
runContext,
224-
HttpRequest
225-
.create(
226-
HttpMethod.GET,
227-
UriTemplate
228-
.of("/api/v2/accounts/{accountId}/runs/{runId}/artifacts/{path}")
229-
.expand(Map.of(
230-
"accountId", runContext.render(this.accountId).as(String.class).orElseThrow(),
231-
"runId", runId,
232-
"path", path
233-
))
234-
),
235-
Argument.of(String.class)
236-
)
237-
.getBody()
238-
.orElseThrow();
203+
private Path downloadArtifacts(RunContext runContext, Long runId, String path) throws IllegalVariableEvaluationException, IOException, HttpClientException {
204+
HttpRequest.HttpRequestBuilder requestBuilder = HttpRequest.builder()
205+
.uri(URI.create(runContext.render(this.baseUrl).as(String.class).orElseThrow() + "/api/v2/accounts/" + runContext.render(this.accountId).as(String.class).orElseThrow() + "/runs/" + runId + "/artifacts/" + path))
206+
.method("GET");
239207

240-
Path tempFile = runContext.workingDir().createTempFile(".json");
208+
String artifact = this.request(runContext, requestBuilder, String.class).getBody();
241209

210+
Path tempFile = runContext.workingDir().createTempFile(".json");
242211
Files.writeString(tempFile, artifact, StandardOpenOption.TRUNCATE_EXISTING);
243212

244213
return tempFile;

‎src/main/java/io/kestra/plugin/dbt/cloud/TriggerRun.java

Lines changed: 17 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,13 @@
11
package io.kestra.plugin.dbt.cloud;
22

3+
import io.kestra.core.http.HttpRequest;
4+
import io.kestra.core.http.HttpResponse;
35
import io.kestra.core.models.annotations.Example;
46
import io.kestra.core.models.annotations.Plugin;
57
import io.kestra.core.models.property.Property;
68
import io.kestra.core.models.tasks.RunnableTask;
79
import io.kestra.core.runners.RunContext;
810
import io.kestra.plugin.dbt.cloud.models.RunResponse;
9-
import io.micronaut.core.type.Argument;
10-
import io.micronaut.http.HttpMethod;
11-
import io.micronaut.http.HttpRequest;
12-
import io.micronaut.http.HttpResponse;
13-
import io.micronaut.http.uri.UriTemplate;
1411
import io.swagger.v3.oas.annotations.media.Schema;
1512
import jakarta.validation.constraints.NotNull;
1613
import lombok.Builder;
@@ -163,23 +160,21 @@ public TriggerRun.Output run(RunContext runContext) throws Exception {
163160
body.put("steps_override", runContext.render(this.stepsOverride).asList(String.class));
164161
}
165162

166-
HttpResponse<RunResponse> triggerResponse = this.request(
167-
runContext,
168-
HttpRequest
169-
.create(
170-
HttpMethod.POST,
171-
UriTemplate
172-
.of("/api/v2/accounts/{accountId}/jobs/{jobId}/run")
173-
.expand(Map.of(
174-
"accountId", runContext.render(this.accountId).as(String.class).orElseThrow(),
175-
"jobId", runContext.render(this.jobId).as(String.class).orElseThrow()
176-
)) + "/"
177-
)
178-
.body(body),
179-
Argument.of(RunResponse.class)
180-
);
181-
182-
RunResponse triggerRunResponse = triggerResponse.getBody().orElseThrow(() -> new IllegalStateException("Missing body on trigger"));
163+
HttpRequest.HttpRequestBuilder requestBuilder = HttpRequest.builder()
164+
.uri(URI.create(runContext.render(this.baseUrl).as(String.class).orElseThrow() + "/api/v2/accounts/" + runContext.render(this.accountId).as(String.class).orElseThrow() +
165+
"/jobs/" + runContext.render(this.jobId).as(String.class).orElseThrow() + "/run/"))
166+
.method("POST")
167+
.body(HttpRequest.JsonRequestBody.builder()
168+
.content(body)
169+
.build());
170+
171+
HttpResponse<RunResponse> triggerResponse = this.request(runContext, requestBuilder, RunResponse.class);
172+
173+
RunResponse triggerRunResponse = triggerResponse.getBody();
174+
if (triggerRunResponse == null) {
175+
throw new IllegalStateException("Missing body on trigger");
176+
}
177+
183178
logger.info("Job status {} with response: {}", triggerResponse.getStatus(), triggerRunResponse);
184179
Long runId = triggerRunResponse.getData().getId();
185180

0 commit comments

Comments
 (0)