FEAT: enhance pipeline run jobs endpoint with error handling and response structure

This commit is contained in:
iruy-fr
2026-06-22 20:25:16 -03:00
parent b5f569e522
commit 63efff6adf
3 changed files with 170 additions and 64 deletions
+32 -46
View File
@@ -3343,7 +3343,14 @@
],
"responses": {
"200": {
"description": ""
"description": "",
"content": {
"application/json": {
"schema": {
"type": "object"
}
}
}
}
},
"tags": [
@@ -4570,50 +4577,6 @@
]
}
},
"/platform/pipelines/{pipelineId}/pipeline_run/{runId}/jobs": {
"get": {
"operationId": "PlatformApiController_getPipelineRunJobs",
"summary": "Get pipeline run jobs",
"parameters": [
{
"name": "pipelineId",
"required": true,
"in": "path",
"schema": {
"type": "string"
}
},
{
"name": "runId",
"required": true,
"in": "path",
"schema": {
"type": "string"
}
}
],
"responses": {
"200": {
"description": "",
"content": {
"application/json": {
"schema": {
"type": "object"
}
}
}
}
},
"tags": [
"Platform API"
],
"security": [
{
"access-token": []
}
]
}
},
"/platform/pipelines/{pipelineId}/pipeline_run/{runId}/jobs": {
"get": {
"operationId": "PlatformApiController_getPipelineRunJobs",
@@ -8647,10 +8610,33 @@
"Health"
]
}
},
"/release_note": {
"get": {
"operationId": "ReleaseNoteController_getLatestReleaseNote",
"parameters": [],
"responses": {
"200": {
"description": "",
"content": {
"application/json": {
"schema": {
"type": "object"
}
}
}
}
},
"security": [
{
"access-token": []
}
]
}
}
},
"info": {
"title": "Maestro",
"title": "Maestro - feat/pipeline-run-jobs",
"description": "This is the Maestro API",
"version": "1.0.0",
"contact": {}
+110 -6
View File
@@ -49,9 +49,23 @@ import { Language } from 'src/decorators/language.decorator';
import { ApiInternalOnlyEndpoint } from 'src/decorators/swagger.decorator';
import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
import { PipelineExecutionGuard } from 'src/guards/pipeline-execution.guard';
import { PlatformApiService } from '../platform-api/platform-api.service';
type PipelineTable = { name: string; job_id?: string; is_deleted?: boolean; [key: string]: any };
type PipelineTablesConfig = { input_id?: string; tables: PipelineTable[] };
type PlatformPipelineJob = { job_id?: string; input?: Record<string, any> };
type PlatformPipeline = {
pipeline_id?: string;
created_at?: string;
description?: string;
name?: string;
last_status?: string;
status?: string;
cron?: string;
jobs?: PlatformPipelineJob[];
properties?: Record<string, any>;
user_id?: string;
};
@ApiTags('PipelinesV2')
@ApiHeaders([{ name: 'dadosfera-lang', enum: LanguageEnum, required: false }])
@@ -63,10 +77,70 @@ export class PipelinesController {
@Inject(DadosferaLogger)
dadosferaLogger: DadosferaLogger,
private pipelinesClientService: PipelinesService,
private platformApiService: PlatformApiService,
) {
this.logger = dadosferaLogger.logger;
}
private normalizePipelineId(id: string): string {
return id?.replace(/-/g, '_') || '';
}
private buildPlatformPipelineFallback(
id: string,
platformPipeline: PlatformPipeline,
): Messages.PipelineV2FindOneResponse {
const jobs = platformPipeline.jobs || [];
const firstInput = jobs.find((job) => job.input)?.input || {};
const plugin = firstInput.plugin || firstInput.connector;
const tables = jobs.map((job) => ({
...job.input,
name:
job.input?.table_name ||
job.input?.source_prefix ||
job.input?.stream ||
job.job_id ||
'',
job_id: job.job_id,
}));
return {
pipeline: {
id,
created_at: platformPipeline.created_at,
description: platformPipeline.description,
name: platformPipeline.name,
transformations: [],
status: platformPipeline.status || platformPipeline.last_status || '',
input: {
...firstInput,
plugin,
category: firstInput.category || (firstInput.connector === 's3' ? 'file' : 'database'),
cron: platformPipeline.cron,
tables,
input_id: firstInput.input_id || null,
},
properties: platformPipeline.properties || {},
username: platformPipeline.user_id,
} as any,
};
}
private async getPipelineFromPlatformApi(
id: string,
user: RequestUser,
): Promise<Messages.PipelineV2FindOneResponse> {
const normalizedId = this.normalizePipelineId(id);
const platformPipeline = await this.platformApiService.proxy(
'GET',
`/pipeline/${normalizedId}`,
user,
);
return this.buildPlatformPipelineFallback(id, platformPipeline);
}
@Get('monitoring-dashboard')
@RequireSomePermission(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getMonitoringDashboard(@User() user: RequestUser) {
@@ -189,18 +263,38 @@ export class PipelinesController {
@Get(':id/status')
@RequireSomePermission(PERMISSIONS_GROUPS.IMPORT_FILES.permissions.VIEW, PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
async getPipelineStatus(@Body() body, @Param('id') id: string) {
body.id = id;
async getPipelineStatus(
@Param('id') id: string,
@User() user: RequestUser,
) {
const body = {
id,
info: {
customer_id: user.customer_id,
user_id: user.user_id,
customer: user.customer_name,
},
};
this.logger.info(`/pipeline/${id} - ON GET PIPELINE STATUS ROUTE`, {
user: body.info.user_id,
customer: body.info.customer,
});
const response = await this.pipelinesClientService.getPipelineStatus(body);
try {
return await this.pipelinesClientService.getPipelineStatus(body);
} catch (error) {
this.logger.warn('PipelinesController - getPipelineStatus fallback to platform-api', {
id,
error: error.message,
});
return response;
return this.platformApiService.proxy(
'GET',
`/pipeline/${this.normalizePipelineId(id)}/pipeline_run`,
user,
);
}
}
@Get('/:id')
@@ -222,7 +316,17 @@ export class PipelinesController {
language,
});
const pipelineRes = await this.pipelinesClientService.findOne({ id }, metadata);
let pipelineRes: Messages.PipelineV2FindOneResponse;
try {
pipelineRes = await this.pipelinesClientService.findOne({ id }, metadata);
} catch (error) {
this.logger.warn('PipelinesController - findOne fallback to platform-api', {
id,
error: error.message,
});
return this.getPipelineFromPlatformApi(id, user);
}
const parsed: PipelineTablesConfig = JSON.parse(pipelineRes.pipeline.config.tables);
const input_id = parsed.input_id;
@@ -72,6 +72,10 @@ export class PlatformApiController {
return id?.replace(/-/g, '_') || '';
}
private decodePathParam(value: string): string {
return value ? decodeURIComponent(value) : '';
}
/**
* Denormalize ID back to UUID format (replace _ with -).
* Used when we receive a normalized ID but need the original UUID.
@@ -778,10 +782,10 @@ export class PlatformApiController {
@User() user: RequestUser,
) {
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
const normalizedRunId = this.normalizePipelineId(runId);
const decodedRunId = this.decodePathParam(runId);
return this.platformApiService.proxy(
'GET',
`/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}`,
`/pipeline/${normalizedPipelineId}/pipeline_run/${decodedRunId}`,
user,
);
}
@@ -794,10 +798,10 @@ export class PlatformApiController {
@User() user: RequestUser,
@Query() query: Record<string, string>,
) {
const normalizedRunId = this.normalizePipelineId(runId);
const decodedRunId = this.decodePathParam(runId);
return this.platformApiService.proxy(
'GET',
`/pipeline/pipeline_run/${normalizedRunId}/logs`,
`/pipeline/pipeline_run/${decodedRunId}/logs`,
user,
undefined,
query,
@@ -813,7 +817,7 @@ export class PlatformApiController {
@User() user: RequestUser,
) {
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
const normalizedRunId = this.normalizePipelineId(runId);
const decodedRunId = this.decodePathParam(runId);
const status = await this.platformApiService.proxy(
'GET',
@@ -827,7 +831,7 @@ export class PlatformApiController {
return this.platformApiService.proxy(
'POST',
`/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}/cancel`,
`/pipeline/${normalizedPipelineId}/pipeline_run/${decodedRunId}/cancel`,
user,
);
}
@@ -857,13 +861,25 @@ export class PlatformApiController {
@User() user: RequestUser,
) {
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
const normalizedRunId = this.normalizePipelineId(runId);
const decodedRunId = this.decodePathParam(runId);
return this.platformApiService.proxy(
'GET',
`/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}/jobs`,
user,
);
try {
return await this.platformApiService.proxy(
'GET',
`/pipeline/${normalizedPipelineId}/pipeline_run/${decodedRunId}/jobs`,
user,
);
} catch (error) {
if (error instanceof HttpException && error.getStatus() === 404) {
this.logger.warn('Platform API pipeline run jobs not found; returning empty jobs list', {
pipelineId: normalizedPipelineId,
runId: decodedRunId,
});
return { jobs: [] };
}
throw error;
}
}
// ==================== JOBS - COLUMN EDITING ROUTES ====================