gitea_admin/maestro
1
0
Fork 0
mirror of https://github.com/dadosfera/maestro.git synced 2026-08-31 19:58:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #476 from dadosfera/feat/beta-delete-pipeline-job

Browse Source 
Feat/beta delete pipeline job
This commit is contained in:
vinicius gadea
2026-04-02 17:35:06 -03:00
committed by GitHub
parent 7b1049224c efb0b58648
commit bf12598915
8 changed files with 132 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docsfera.json
+37
View File
@@ -4705,6 +4705,43 @@
]
}
},
"/platform/pipelines/{pipelineId}/inputs/{inputId}": {
"delete": {
"operationId": "PlatformApiController_deleteTable",
"summary": "Mark a table as deleted and delete its associated job via platform-api",
"parameters": [
{
"name": "pipelineId",
"required": true,
"in": "path",
"schema": {
"type": "string"
}
},
{
"name": "inputId",
"required": true,
"in": "path",
"schema": {
"type": "string"
}
}
],
"responses": {
"200": {
"description": ""
}
},
"tags": [
"Platform API"
],
"security": [
{
"access-token": []
}
]
}
},
"/platform/jobs/jdbc/{jobId}": {
"get": {
"operationId": "PlatformApiController_getJdbcJob",
package-lock.json
Generated
+4 -5
View File
@@ -17,7 +17,7 @@
"@aws-sdk/signature-v4": "^3.370.0",
"@dadosfera/dadosfera-logs": "^1.0.0-beta.4",
"@dadosfera/protospack": "2.5.3",
"@dadosfera/protospack-v2": "3.40.0-beta.2",
"@dadosfera/protospack-v2": "^3.40.0-beta.5",
"@grpc/grpc-js": "^1.9.3",
"@grpc/proto-loader": "^0.7.9",
"@nestjs/cli": "^9.5.0",
@@ -1745,10 +1745,9 @@
}
},
"node_modules/@dadosfera/protospack-v2": {
"version": "3.40.0-beta.2",
"resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.2.tgz",
"integrity": "sha512-WMoL9OhKJ05qDojd7cPcP7x45bHHnnQttfVoTWrc+lOkcVHygu0ob52ArfD8RRmpAXezBh9vDSbxz45n/lxbKg==",
"license": "ISC",
"version": "3.40.0-beta.5",
"resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.5.tgz",
"integrity": "sha512-iocKv/XXp2jKAasO5ONgm31cKfLgNsU4pEKZMr6YnR7nQaH11WcW7rnuagNxWoik++wLUqbYyf0bZWRDzMlCPA==",
"dependencies": {
"@grpc/grpc-js": "^1.9.3",
"rxjs": "^7.5.5"
package.json
+1 -1
View File
@@ -35,7 +35,7 @@
"@aws-sdk/signature-v4": "^3.370.0",
"@dadosfera/dadosfera-logs": "^1.0.0-beta.4",
"@dadosfera/protospack": "2.5.3",
"@dadosfera/protospack-v2": "3.40.0-beta.2",
"@dadosfera/protospack-v2": "^3.40.0-beta.5",
"@grpc/grpc-js": "^1.9.3",
"@grpc/proto-loader": "^0.7.9",
"@nestjs/cli": "^9.5.0",
src/modules/inputs/inputs.service.ts
+8
View File
@@ -288,4 +288,12 @@ export class InputsService {
};
return formatedPayload;
}
async markTableDeleted(data: { input_id: string; table_name: string; info: Info }) {
return lastValueFrom(this.inputWriteService.MarkTableDeleted(data));
}
async unmarkTableDeleted(data: { input_id: string; table_name: string; info: Info }) {
return lastValueFrom((this.inputWriteService as any).UnmarkTableDeleted(data));
}
}
src/modules/pipelinesV2/pipelines.controller.ts
+24 -23
View File
@@ -53,6 +53,9 @@ import { TableColumns } from '../inputs/dtos/input.model';
import { UpdateInputRequest } from '../inputs/dtos/old_interfaces';
import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
type PipelineTable = { name: string; job_id?: string; is_deleted?: boolean; [key: string]: any };
type PipelineTablesConfig = { input_id?: string; tables: PipelineTable[] };
@ApiTags('PipelinesV2')
@ApiHeaders([{ name: 'dadosfera-lang', enum: LanguageEnum, required: false }])
@UseFilters(new GrpcToHttpExceptionFilter())
@@ -62,6 +65,7 @@ export class PipelinesController {
constructor(
@Inject(DadosferaLogger)
dadosferaLogger: DadosferaLogger,
private pipelinesClientService: PipelinesService,
private oldPipelinesService: OldPipelineService,
) {
@@ -222,30 +226,27 @@ export class PipelinesController {
language,
});
const result = await this.pipelinesClientService
.findOne({ id }, metadata)
.then((res) => {
//{pipeline:{tables: {tables: [], input_id: ''}}}
let tables = JSON.parse(res.pipeline.config.tables);
const input_id = tables?.input_id;
if (tables?.tables) tables = tables.tables;
Object.assign(res.pipeline, {
transformations: res.pipeline.transformations
? JSON.parse(res.pipeline.transformations)
: [],
config: {
cron: res.pipeline.config.cron,
tables,
input_id
},
properties: res.pipeline.properties
? JSON.parse(res.pipeline.properties)
: {},
});
return res;
});
const pipelineRes = await this.pipelinesClientService.findOne({ id }, metadata);
return result;
const parsed: PipelineTablesConfig = JSON.parse(pipelineRes.pipeline.config.tables);
const input_id = parsed.input_id;
const tables: PipelineTable[] = parsed.tables ?? [];
Object.assign(pipelineRes.pipeline, {
transformations: pipelineRes.pipeline.transformations
? JSON.parse(pipelineRes.pipeline.transformations)
: [],
config: {
cron: pipelineRes.pipeline.config.cron,
tables,
input_id,
},
properties: pipelineRes.pipeline.properties
? JSON.parse(pipelineRes.pipeline.properties)
: {},
});
return pipelineRes;
}
@Patch('/:id')
src/modules/platform-api/platform-api.controller.ts
+48
View File
@@ -11,6 +11,7 @@ import {
Inject,
BadRequestException,
HttpException,
NotFoundException,
} from '@nestjs/common';
import { ApiTags, ApiOperation } from '@nestjs/swagger';
import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
@@ -29,6 +30,7 @@ import { validateCronAgainstScheduleLimit } from '../../utils/cron-validation';
import { CatalogService } from '../catalog/catalog.service';
import { PackTheMetadata } from '../../utils/PackTheMetadata';
import { ValidationTableDTO } from './platform-api.dto';
import { InputsService } from '../inputs/inputs.service';
type ValidateTablesDTO = {
@@ -54,6 +56,7 @@ export class PlatformApiController {
private readonly dynamoDBService: DynamoDBService,
private readonly customersService: CustomersService,
private readonly catalogService: CatalogService,
private readonly inputsService: InputsService,
@Inject(DadosferaLogger) dadosferaLogger: DadosferaLogger,
) {
this.logger = dadosferaLogger.logger;
@@ -962,6 +965,51 @@ export class PlatformApiController {
);
}
@Delete('pipelines/:pipelineId/inputs/:inputId')
@ApiOperation({ summary: 'Mark a table as deleted and delete its associated job via platform-api' })
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.DELETE)
async deleteTable(
@Param('pipelineId') pipelineId: string,
@Param('inputId') inputId: string,
@Body() body: { table_name: string },
@User() user: RequestUser,
) {
const tableName = body.table_name;
const info = {
customer_id: user.customer_id,
customer: user.customer_name,
user_id: user.user_id,
};
this.logger.info('deleteTable: marking table as deleted', { inputId, tableName });
const updatedInput: any = await this.inputsService.markTableDeleted({ input_id: inputId, table_name: tableName, info });
this.logger.info('deleteTable: table marked as deleted', { inputId, tableName });
try {
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
this.logger.info('deleteTable: fetching pipeline from platform-api', { pipelineId, normalizedPipelineId });
const platformPipeline = await this.platformApiService.proxy('GET', `/pipeline/${normalizedPipelineId}`, user);
this.logger.info('deleteTable: pipeline fetched', { jobCount: platformPipeline?.jobs?.length });
const job = platformPipeline?.jobs?.find((j: any) => j.input?.table_name === tableName);
if (!job) throw new NotFoundException(`Job for table '${tableName}' not found in pipeline`);
this.logger.info('deleteTable: deleting job from platform-api', { jobId: job.job_id });
await this.platformApiService.proxy('DELETE', `/jobs/${job.job_id}`, user);
this.logger.info('deleteTable: job deleted', { jobId: job.job_id });
return { name: tableName, is_deleted: updatedInput.is_deleted ?? true, deleted_at: updatedInput.deleted_at };
} catch (error) {
this.logger.error('deleteTable: platform-api delete failed, attempting rollback', { tableName, error: error.message });
try {
await this.inputsService.unmarkTableDeleted({ input_id: inputId, table_name: tableName, info });
} catch (rollbackError) {
this.logger.error('deleteTable: rollback failed', { tableName, error: rollbackError.message });
}
throw error;
}
}
// ==================== JOBS - JDBC SYNC MODE ROUTES ====================
@Get('jobs/jdbc/:jobId')
src/modules/platform-api/platform-api.module.ts
+2 -1
View File
@@ -8,9 +8,10 @@ import { ElasticsearchModule } from '../../services/elasticsearch';
import { DynamoDBModule } from '../../services/dynamodb';
import { CustomersModule } from '../customers/customers.module';
import { CatalogModule } from '../catalog/catalog.module';
import { InputsModule } from '../inputs/inputs.module';
@Module({
imports: [ElasticsearchModule, DynamoDBModule, CustomersModule, CatalogModule],
imports: [ElasticsearchModule, DynamoDBModule, CustomersModule, CatalogModule, InputsModule],
controllers: [PlatformApiController],
providers: [PlatformApiService, DadosferaLogger],
exports: [PlatformApiService],
src/services/dynamodb/dynamodb.service.ts
+8 -4
View File
@@ -125,9 +125,13 @@ export class DynamoDBService {
},
});
const { Item } = await this.documentClient.send(getCommand);
return Item as InputDocument | null;
try {
const { Item } = await this.documentClient.send(getCommand);
return Item as InputDocument | null;
} catch (error) {
this.logger.error('DynamoDB: findInput failed', { inputId, clientId, error: error.message });
throw error;
}
}
async deleteInput(clientId: string, inputId: string): Promise<void> {
@@ -203,7 +207,6 @@ export class DynamoDBService {
updatedTable.reference_column = changes.reference_column;
}
}
tables[tableIndex] = updatedTable;
// Save updated document
@@ -231,4 +234,5 @@ export class DynamoDBService {
throw error;
}
}
}