FIX: pipeline delete working

This commit is contained in:
Gabriel Rosa
2022-10-25 16:21:58 -03:00
parent 526a3bddcd
commit 4ffd0916c3
11 changed files with 102 additions and 77 deletions
+1 -15
View File
@@ -91,14 +91,6 @@ export class InputsService {
return updateInputResponse;
},
remove: async (idRequest: IIdRequest) => {
this.logger.info('InputClientService - Remove');
const removeInputResponse = await lastValueFrom(
this.inputWriteService.InputRemove(idRequest),
);
return removeInputResponse;
},
testConnection: async (data: TestConnectionRequest) => {
this.logger.info('InputClientService - TestConnection');
const testConnectionResponse = await lastValueFrom(
@@ -271,13 +263,7 @@ export class InputsService {
}
async remove(idRequest: IIdRequest) {
try {
const removeInputResponse = await this.OLD_inputClient.remove(idRequest);
return removeInputResponse;
} catch (err) {
throw new HttpException(err.message, HttpStatus.NOT_FOUND);
}
return lastValueFrom(this.inputWriteService.InputRemove(idRequest));
}
@Timeout(60000 * 10) // Timeout set for 10 minutes
@@ -1,7 +1,6 @@
import {
Body,
Controller,
Delete,
Get,
Inject,
Param,
@@ -156,14 +155,4 @@ export class PipelinesController {
return response;
}
@Delete(':id')
async delete(@Body() data, @Param() params) {
const { id } = params;
this.logger.info(process.env.DEV_URL + `/pipeline/${id} ON DELETE ROUTE`);
const response = await this.pipelineService.remove({ id, ...data });
return response;
}
}
@@ -73,16 +73,6 @@ export class PipelinesService {
}
}
async remove(data: IIdRequest) {
try {
const removePipelineResponse = await this.pipelineClient.remove(data);
return removePipelineResponse;
} catch (err) {
throw new HttpException(err.message, HttpStatus.NOT_FOUND);
}
}
async getPipelineLogsMessages(id: string, details: string) {
try {
const pipelineLogsResponse =
@@ -9,8 +9,15 @@ import {
Put,
Headers,
Query,
UseFilters,
HttpCode,
HttpStatus,
} from '@nestjs/common';
import { ApiCreatedResponse, ApiTags } from '@nestjs/swagger';
import {
ApiCreatedResponse,
ApiNoContentResponse,
ApiTags,
} from '@nestjs/swagger';
import { AuthenticateCondition } from 'src/authentication/authentication.decorator';
import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
import { PipelinesService } from './pipelines.service';
@@ -21,7 +28,10 @@ import { PackTheMetadata } from 'src/utils/ PackTheMetadata';
import { PipelinesService as OldPipelineService } from 'src/modules/pipelines/pipelines.service';
import { ICreatePipelineV2Req, IPipelineV2 } from './interfaces';
@ApiTags('Pipelines')
import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filter';
@UseFilters(new GrpcToHttpExceptionFilter())
@ApiTags('PipelinesV2')
@Controller('pipelinesV2')
@AuthenticateCondition((req, user) => {
let action;
@@ -143,13 +153,16 @@ export class PipelinesController {
const result = await this.pipelinesClientService
.findOne({ id }, metadata)
.then((res) => {
//{pipeline:{tables: {tables: [], input_id: ''}}}
let tables = JSON.parse(res.pipeline.config.tables);
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: JSON.parse(res.pipeline.config.tables),
tables,
},
properties: res.pipeline.properties
? JSON.parse(res.pipeline.properties)
@@ -159,6 +172,7 @@ export class PipelinesController {
});
return result;
}
@Put('/:id')
async update(
@Body() updatePipelineDto,
@@ -179,12 +193,17 @@ export class PipelinesController {
}
@Delete(':id')
async delete(@Body() data, @Param() params, @User() user: RequestUser) {
@ApiNoContentResponse()
@HttpCode(HttpStatus.NO_CONTENT)
async delete(@Param() params, @User() user: RequestUser) {
this.logger.info('PipelinesController - delete', { user });
const { id } = params;
this.logger.info('PipelinesController - findOne', { user });
const response = await this.pipelinesClientService.remove({ id, ...data });
return response;
const metadata = PackTheMetadata({
customer_id: user.customer_id,
customer_name: user.customer_name,
user_id: user.user_id,
});
await this.pipelinesClientService.remove({ id, metadata, user });
this.logger.info('PipelinesController - delete: OK');
}
}
@@ -9,6 +9,8 @@ import { PipelinesClientConfiguration } from './pipelines-client';
import { PipelinesModule as OldPipelineModule } from 'src/modules/pipelines/pipelines.module';
import { ConnectorModule } from '../connector/connector.module';
import { InputsModule } from '../inputs/inputs.module';
import { TransformationsModule } from '../transformations/transformations.module';
const client = new PipelinesClientConfiguration();
@@ -17,6 +19,8 @@ const client = new PipelinesClientConfiguration();
ClientsModule.register([client.providerOptions]),
OldPipelineModule,
ConnectorModule,
InputsModule,
TransformationsModule,
],
controllers: [PipelinesController],
providers: [PipelinesService, DadosferaLogger],
+39 -26
View File
@@ -19,6 +19,9 @@ import { ICreatePipelineV2Req } from './interfaces';
import { PipelineV2CreateRequest } from '@dadosfera/protospack-v2/dist/lib/PipelineV2/interfaces/messages';
import { Metadata } from '@grpc/grpc-js';
import { ConnectorClientService } from '../connector/client.service';
import { InputsService } from '../inputs/inputs.service';
import { RequestUser } from 'src/authentication/user.decorator';
import { TransformationsService } from '../transformations/transformations.service';
export class PipelinesService implements OnModuleInit {
logger: DadosferaLogger;
@@ -30,6 +33,8 @@ export class PipelinesService implements OnModuleInit {
@Inject(PipelinesClientConfiguration.name)
private readonly grpcClient: ClientGrpc,
private readonly connectorService: ConnectorClientService,
private readonly inputsService: InputsService,
private readonly transformationsService: TransformationsService,
) {
this.logger = dadosferaLogger.logger;
}
@@ -128,10 +133,7 @@ export class PipelinesService implements OnModuleInit {
const findOnePipelineResponse = await lastValueFrom(
this.pipelineReadService.PipelineV2FindOne(data, metadata),
).catch((err) => {
this.logger.error(err.message);
throw new Error(err);
});
);
this.logger.info('Done');
return findOnePipelineResponse;
@@ -142,35 +144,46 @@ export class PipelinesService implements OnModuleInit {
const updatePipelineResponse = await lastValueFrom(
this.pipelineWriteService.PipelineV2Update(UpdatePipelineRequest),
)
.then((res) => {
this.logger.info('Done');
return res;
})
.catch((err) => {
this.logger.error(err.message);
throw new Error(err);
});
);
this.logger.info('Done');
return updatePipelineResponse;
}
async remove(data: Messages.PipelineV2RemoveRequest) {
this.logger.info('PipelinesClientService - Remove');
async remove(data: { id: string; metadata: Metadata; user: RequestUser }) {
const { id, metadata, user } = data;
const info = {
user_id: user.user_id,
customer_id: user.customer_id,
customer: user.customer_name,
};
const { pipeline } = await lastValueFrom(
this.pipelineReadService.PipelineV2FindOne({ id }, metadata),
);
await lastValueFrom(
this.pipelineWriteService.PipelineV2Remove({ id }, metadata),
);
const removePipelineResponse = await lastValueFrom(
this.pipelineWriteService.PipelineV2Remove(data),
)
.then((res) => {
this.logger.info('Done');
return res;
})
.catch((err) => {
this.logger.error(err.message);
throw new Error(err);
//{pipeline:{tables: {tables: [], input_id: ''}}}
const input = pipeline.config.tables
? JSON.parse(pipeline.config.tables)
: null;
if (input)
await this.inputsService.remove({
id: input.input_id,
info,
});
return removePipelineResponse;
const transformations: { id: string }[] = pipeline.transformations
? JSON.parse(pipeline.transformations)
: null;
if (transformations && transformations.length)
for (const transformation of transformations) {
await this.transformationsService.remove({
id: transformation.id,
info,
});
}
}
async findOneProperties(id: string, metadata: Metadata) {