fix: add get_root_job_id

This commit is contained in:
Ruben Fiszel
2024-02-06 00:17:50 +01:00
parent 6065506441
commit 7db2501cc1
5 changed files with 98 additions and 38 deletions
+17
View File
@@ -4980,6 +4980,23 @@ paths:
schema:
$ref: "#/components/schemas/Job"
/w/{workspace}/jobs_u/get_root_job_id/{id}:
get:
summary: get root job id
operationId: getRootJobId
tags:
- job
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/JobId"
responses:
"200":
description: get root job id
content:
application/json:
schema:
type: string
/w/{workspace}/jobs_u/get_logs/{id}:
get:
summary: get job logs
+21
View File
@@ -204,6 +204,7 @@ pub fn global_service() -> Router {
"/get_flow/:job_id/:resume_id/:secret",
get(get_suspended_job_flow),
)
.route("/get_root_job_id/:id", get(get_root_job))
.route("/get/:id", get(get_job))
.route("/get_logs/:id", get(get_job_logs))
.route("/get_flow_debug_info/:id", get(get_flow_job_debug_info))
@@ -239,6 +240,26 @@ async fn get_result_by_id(
Ok(Json(res))
}
async fn get_root_job(
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
) -> windmill_common::error::JsonResult<String> {
let res = compute_root_job_for_flow(&db, &w_id, id).await?;
Ok(Json(res))
}
async fn compute_root_job_for_flow(db: &DB, w_id: &str, job_id: Uuid) -> error::Result<String> {
let mut job = get_queued_job(job_id, w_id, db).await?;
while let Some(j) = job {
if let Some(uuid) = j.parent_job {
job = get_queued_job(uuid, w_id, db).await?;
} else {
return Ok(j.id.to_string());
}
}
Ok(job_id.to_string())
}
async fn get_db_clock(Extension(db): Extension<DB>) -> windmill_common::error::JsonResult<i64> {
Ok(Json(now_from_db(&db).await?.timestamp_millis()))
}
@@ -108,7 +108,11 @@
}
if (keys.length != 0) {
if (keys.length == 1 && keys[0] == 'table-row') {
if (Array.isArray(result) && result.every((elt) => inferResultKind(elt) === 's3object')) {
return 's3object-list'
} else if (isRectangularArray(result)) {
return 'table-col'
} else if (keys.length == 1 && keys[0] == 'table-row') {
return 'table-row'
} else if (
(keys.length == 1 && keys[0] == 'table-col') ||
@@ -140,7 +144,6 @@
a.href = 'data:application/octet-stream;base64,' + result.file
a.download = result.filename
a.click()
console.log('autodownload', result.file, result.filename)
}
return 'file'
} else if (
@@ -154,13 +157,6 @@
} else if (keys.length === 1 && (keys.includes('md') || keys.includes('markdown'))) {
return 'markdown'
}
} else if (
Array.isArray(result) &&
result.every((elt) => inferResultKind(elt) === 's3object')
) {
return 's3object-list'
} else if (isRectangularArray(result)) {
return 'table-col'
}
} catch (err) {}
} else {
+9
View File
@@ -221,6 +221,11 @@ class Windmill:
def get_job(self, job_id: str) -> dict:
return self.get(f"/w/{self.workspace}/jobs_u/get/{job_id}").json()
def get_root_job_id(self, job_id: str | None = None) -> dict:
job_id = job_id or os.environ.get("WM_JOB_ID")
return self.get(f"/w/{self.workspace}/jobs_u/get_root_job_id/{job_id}").json()
def get_id_token(self, audience: str) -> dict:
return self.post(f"/w/{self.workspace}/oidc/token/{audience}").text
@@ -572,6 +577,10 @@ def deprecate(in_favor_of: str):
def get_workspace() -> str:
return _client.workspace
@init_global_client
def get_root_job_id(job_id: str | None = None) -> str:
return _client.get_root_job_id(job_id)
@init_global_client
@deprecate("Windmill().version")
+46 -29
View File
@@ -95,6 +95,18 @@ export async function getResource(
}
}
/**
* Get a resource value by path
* @param jobId job id to get the root job id from (default to current job)
* @returns root job id
*/
export async function getRootJobId(jobId?: string): Promise<string> {
!clientSet && setClient();
const workspace = getWorkspace();
jobId = jobId ?? getEnv("WM_JOB_ID");
return await JobService.getRootJobId({ workspace, id: jobId });
}
/**
* Resolve a resource value in case the default value was picked because the input payload was undefined
* @param obj resource value or path of the resource under the format `$res:path`
@@ -295,7 +307,7 @@ export async function denoS3LightClientSettings(
/**
* Load the content of a file stored in S3. If the s3ResourcePath is undefined, it will default to the workspace S3 resource.
*
*
* ```typescript
* let fileContent = await wmill.loadS3FileContent(inputFile)
* // if the file is a raw text file, it can be decoded and printed directly:
@@ -306,39 +318,39 @@ export async function denoS3LightClientSettings(
export async function loadS3File(
s3object: S3Object,
s3ResourcePath: string | undefined
): Promise<Uint8Array|undefined> {
): Promise<Uint8Array | undefined> {
!clientSet && setClient();
const fileContentBlob = await loadS3FileStream(s3object, s3ResourcePath)
const fileContentBlob = await loadS3FileStream(s3object, s3ResourcePath);
if (fileContentBlob === undefined) {
return undefined
return undefined;
}
// we read the stream until completion and put the content in an Uint8Array
const reader = fileContentBlob.stream().getReader()
const reader = fileContentBlob.stream().getReader();
const chunks: Uint8Array[] = [];
while (true) {
const {value: chunk, done} = await reader.read();
const { value: chunk, done } = await reader.read();
if (done) {
break;
}
chunks.push(chunk);
}
let fileContentLength = 0;
chunks.forEach(item => {
chunks.forEach((item) => {
fileContentLength += item.length;
});
let fileContent = new Uint8Array(fileContentLength);
let offset = 0;
chunks.forEach(chunk => {
chunks.forEach((chunk) => {
fileContent.set(chunk, offset);
offset += chunk.length;
});
return fileContent
return fileContent;
}
/**
* Load the content of a file stored in S3 as a stream. If the s3ResourcePath is undefined, it will default to the workspace S3 resource.
*
*
* ```typescript
* let fileContentBlob = await wmill.loadS3FileStream(inputFile)
* // if the content is plain text, the blob can be read directly:
@@ -348,29 +360,34 @@ export async function loadS3File(
export async function loadS3FileStream(
s3object: S3Object,
s3ResourcePath: string | undefined
): Promise<Blob|undefined> {
): Promise<Blob | undefined> {
!clientSet && setClient();
let params: Record<string, string> = {}
params["file_key"] = s3object.s3
let params: Record<string, string> = {};
params["file_key"] = s3object.s3;
if (s3ResourcePath !== undefined) {
params["s3_resource_path"] = s3ResourcePath
params["s3_resource_path"] = s3ResourcePath;
}
const queryParams = new URLSearchParams(params);
// We use raw fetch here b/c OpenAPI generated client doesn't handle Blobs nicely
const fileContentBlob = await fetch(`${OpenAPI.BASE}/w/${getWorkspace()}/job_helpers/download_s3_file?${queryParams}`, {
method: 'GET',
headers: {
'Authorization': `Bearer ${OpenAPI.TOKEN}`,
},
})
return fileContentBlob.blob()
const fileContentBlob = await fetch(
`${
OpenAPI.BASE
}/w/${getWorkspace()}/job_helpers/download_s3_file?${queryParams}`,
{
method: "GET",
headers: {
Authorization: `Bearer ${OpenAPI.TOKEN}`,
},
}
);
return fileContentBlob.blob();
}
/**
* Persist a file to the S3 bucket. If the s3ResourcePath is undefined, it will default to the workspace S3 resource.
*
*
* ```typescript
* const s3object = await writeS3File(s3Object, "Hello Windmill!")
* const fileContentAsUtf8Str = (await s3object.toArray()).toString('utf-8')
@@ -383,13 +400,13 @@ export async function writeS3File(
s3ResourcePath: string | undefined
): Promise<S3Object> {
!clientSet && setClient();
let fileContentBlob: Blob
if (typeof fileContent === 'string') {
let fileContentBlob: Blob;
if (typeof fileContent === "string") {
fileContentBlob = new Blob([fileContent as string], {
type: 'text/plain'
type: "text/plain",
});
} else {
fileContentBlob = fileContent as Blob
fileContentBlob = fileContent as Blob;
}
const response = await HelpersService.fileUpload({
@@ -398,10 +415,10 @@ export async function writeS3File(
fileExtension: undefined,
s3ResourcePath: s3ResourcePath,
requestBody: fileContentBlob,
})
});
return {
s3: response.file_key
}
s3: response.file_key,
};
}
/**