gitea_admin/maestro
1
0
Fork 0
mirror of https://github.com/dadosfera/maestro.git synced 2026-11-12 20:59:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

FEAT: inputs and transformation using protospack-v2

Browse Source
This commit is contained in:
Gabriel Rosa
2022-08-16 10:18:54 -03:00
parent 5f17a7106f
commit 6807b3e1a9
9 changed files with 343 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/inputs/dtos/old_interfaces.ts
+1 -19
View File
@@ -11,25 +11,6 @@ interface Values {
schema: string;
}
interface Cron {
hours: string[];
hour_interval: boolean;
hour_resourse: boolean;
hour_resourse_value: number;
month_day: string[];
month_day_interval: boolean;
month_day_resourse: boolean;
month_day_resourse_value: number;
month: string[];
month_interval: boolean;
month_resourse: boolean;
month_resourse_value: number;
week_days: string[];
week_days_interval: boolean;
week_days_resourse: boolean;
week_days_resourse_value: number;
}
export interface ICreateInputRequest {
cron: string;
name: string;
@@ -51,6 +32,7 @@ export interface UpdateInputRequest {
values: Values;
operation: string;
info: Info;
tables: any[];
}
export interface ITestConnectionRequest {
src/modules/inputs/inputs-client.config.ts
+12 -6
View File
@@ -1,24 +1,30 @@
import { Input } from '@dadosfera/protospack-v2';
import { credentials } from '@grpc/grpc-js';
import {
ClientProviderOptions,
GrpcOptions,
Transport,
} from '@nestjs/microservices';
import { InputPackages, InputProtoFilePath } from 'protospack';
export class InputsGrpcClient {
public name = 'InputsGrpcClient';
public readonly name = 'InputsGrpcClient';
private config: GrpcOptions = {
transport: Transport.GRPC,
options: {
url: process.env.INFACTORY_URL,
package: InputPackages,
package: [
Input.ProtoPackages.WritePackage,
Input.ProtoPackages.ReadPackage,
],
credentials: process.env.LOCAL_ENV ? undefined : credentials.createSsl(),
protoPath: InputProtoFilePath,
protoPath: [
Input.ProtoPaths.WriteFilePath,
Input.ProtoPaths.ReadFilePath,
],
loader: {
keepCase: true,
enums: String,
objects: true,
arrays: true,
defaults: false,
},
},
};
src/modules/inputs/inputs.controller.ts
+11 -7
View File
@@ -24,7 +24,8 @@ import {
TestConnectionRes,
} from './dtos/input.model';
import { UpdateInputRequest } from './dtos/old_interfaces';
import { User } from 'src/authentication/user.decorator';
import { RequestUser, User } from 'src/authentication/user.decorator';
import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
@ApiTags('inputs')
@Controller('inputs')
@@ -132,15 +133,18 @@ export class InputsController {
}
@Get()
async findAll(@User() user, @Body() body) {
async findAll(@User() user: RequestUser) {
const { user_id, customer_id, customer_name: customer } = user;
this.logger.info(`/input - ON FIND ALL ROUTE`, {
user: user.user_id,
customer: user.customer_id,
});
const response = await this.inputService.findAll({
info: { user: user.user_id, customer: user.customer_id },
customer,
});
const info: Info = {
customer,
customer_id,
user_id,
};
const response = await this.inputService.findAll(info);
return response;
}
src/modules/inputs/inputs.service.ts
+134 -121
View File
@@ -5,44 +5,56 @@ import {
HttpStatus,
Inject,
Injectable,
InternalServerErrorException,
NotFoundException,
} from '@nestjs/common';
import { Timeout } from '@nestjs/schedule';
import {
DecodeGrpcStruct,
EncodeJsonToGrpcStruct,
InputCreateS3Request,
InputNewCreateRequest,
InputService,
InputServicesNames,
TestConnectionGetColumnsRequest,
TestConnectionRequest,
} from 'protospack';
import CronParser, { CronExpression } from 'cron-parser';
import { Info } from 'protospack/dist/lib/interfaces';
import { ClientGrpc } from '@nestjs/microservices';
import DadosferaLogger from '@dadosfera/dadosfera-logs/dist';
import { lastValueFrom } from 'rxjs';
import {
objectSnakeToCamel,
objectCamelToSnake,
} from 'src/utils/CaseConverter';
import { objectCamelToSnake } from 'src/utils/CaseConverter';
import { IIdRequest, UpdateInputRequest } from './dtos/old_interfaces';
import { InputsGrpcClient } from './inputs-client.config';
import { Input } from '@dadosfera/protospack-v2';
import {
InputCreateS3Request,
InputNewCreateRequest,
TestConnectionGetColumnsRequest,
TestConnectionRequest,
} from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages';
import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
@Injectable()
//TODO refactor this class to remove OLD_inputClient
export class InputsService {
logger: DadosferaLogger;
inputService: InputService;
inputWriteService: Input.WriteService.InputWriteService;
inputReadService: Input.ReadService.InputReadService;
constructor(
@Inject(DadosferaLogger)
dadosferaLogger: DadosferaLogger,
@Inject('InputsGrpcClient') private readonly grpcClient: ClientGrpc,
) {
this.logger = dadosferaLogger.logger;
}
async onModuleInit() {
this.inputWriteService =
this.grpcClient.getService<Input.WriteService.InputWriteService>(
Input.ProtoServices.InputWriteService,
);
this.inputReadService =
this.grpcClient.getService<Input.ReadService.InputReadService>(
Input.ProtoServices.InputReadService,
);
}
OLD_inputClient = {
newCreate: async (createInputDto: InputNewCreateRequest) => {
this.logger.info('InputClientService - Create');
const input = await lastValueFrom(
this.inputService.NewCreate(objectSnakeToCamel(createInputDto)),
this.inputWriteService.InputNewCreate(createInputDto),
);
objectCamelToSnake(input);
return input;
@@ -50,31 +62,30 @@ export class InputsService {
createGeneric: async (createInputGeneric) => {
this.logger.info('InputClientService - Create');
const { info, ...data } = createInputGeneric;
if (data.options) delete data.options;
if (data.credentials.oauth_code) delete data.credentials.oauth_code;
const grpcPayload = {
input: EncodeJsonToGrpcStruct(data),
info,
};
objectSnakeToCamel(grpcPayload);
const structReturn = await lastValueFrom(
this.inputService.Create(grpcPayload),
).catch((err: { details: string }) => {
if (err.details === 'Item Not found!') {
throw new NotFoundException('Input not found');
}
throw new InternalServerErrorException(err.details);
});
objectCamelToSnake(structReturn);
const inputCreated = DecodeGrpcStruct(structReturn.input);
return { input: inputCreated };
if (data.credentials) data.credentials = JSON.stringify(data.credentials);
const input = await lastValueFrom(
this.inputWriteService.InputCreate(data),
);
// .catch((err: { details: string }) => {
// if (err.details === 'Item Not found!') {
// throw new NotFoundException('Input not found');
// }
// throw new InternalServerErrorException(err.details);
// });
// objectCamelToSnake(structReturn);
// const inputCreated = DecodeGrpcStruct(structReturn.input);
return { input };
},
createS3Inputs: async (createInputDto: InputCreateS3Request) => {
this.logger.info('InputClientService - Create');
objectSnakeToCamel(createInputDto);
createInputDto;
const createInputResponse = await lastValueFrom(
this.inputService.CreateS3(createInputDto),
this.inputWriteService.InputCreateS3(createInputDto),
).catch((e) => {
throw new HttpException(e.details, 500);
});
@@ -85,7 +96,7 @@ export class InputsService {
findOne: async (data: IIdRequest) => {
this.logger.info('InputClientService - FindOne');
const findOneInputResponse = await lastValueFrom(
this.inputService.FindOne(objectSnakeToCamel(data)),
this.inputReadService.InputFindOne(data),
).catch((e) => {
this.logger.error(e.details);
throw new HttpException(e.details, HttpStatus.INTERNAL_SERVER_ERROR);
@@ -93,41 +104,32 @@ export class InputsService {
objectCamelToSnake(findOneInputResponse);
return findOneInputResponse;
},
findAll: async (data) => {
this.logger.info('InputClientService - FindAll');
const findAllInputResponse = await lastValueFrom(
this.inputService.FindAll(objectSnakeToCamel(data)),
).catch((e) => {
this.logger.error(e.details);
throw new HttpException(e.details, HttpStatus.INTERNAL_SERVER_ERROR);
});
findAllInputResponse.inputs.forEach((input) => objectCamelToSnake(input));
return findAllInputResponse;
},
update: async (updateInputDTO: UpdateInputRequest) => {
this.logger.info('InputClientService - Update');
const updateInputResponse = await new Promise((resolve, reject) => {
this.inputService.Update(objectSnakeToCamel(updateInputDTO)).subscribe({
next(x) {
resolve(objectCamelToSnake(x));
},
error(err) {
reject(err);
},
complete() {
// console.log('done');
},
});
})
.then((res) => {
this.logger.info('Done');
return res;
})
.catch((err) => {
this.logger.error(err.message);
throw new Error(err);
});
const updateInputResponse = await lastValueFrom(
this.inputWriteService.InputUpdate(updateInputDTO),
);
// const updateInputResponse = await new Promise((resolve, reject) => {
// this.inputWriteService.InputUpdate(updateInputDTO).subscribe({
// next(x) {
// resolve(objectCamelToSnake(x));
// },
// error(err) {
// reject(err);
// },
// complete() {
// // console.log('done');
// },
// });
// })
// .then((res) => {
// this.logger.info('Done');
// return res;
// })
// .catch((err) => {
// this.logger.error(err.message);
// throw new Error(err);
// });
return updateInputResponse;
},
@@ -135,7 +137,7 @@ export class InputsService {
this.logger.info('InputClientService - Remove');
const removeInputResponse = await new Promise((resolve, reject) => {
this.inputService.Remove(objectSnakeToCamel(idRequest)).subscribe({
this.inputWriteService.InputRemove(idRequest).subscribe({
next(x) {
resolve(objectCamelToSnake(x));
},
@@ -161,19 +163,17 @@ export class InputsService {
testConnection: async (data: TestConnectionRequest) => {
this.logger.info('InputClientService - TestConnection');
const testConnectionResponse = await new Promise((resolve, reject) => {
this.inputService
.NewTestConnection(objectSnakeToCamel(data))
.subscribe({
next(x) {
resolve(objectCamelToSnake(x));
},
error(err) {
reject(err);
},
complete() {
// console.log('done');
},
});
this.inputReadService.NewTestConnection(data).subscribe({
next(x) {
resolve(objectCamelToSnake(x));
},
error(err) {
reject(err);
},
complete() {
// console.log('done');
},
});
})
.then((res) => {
this.logger.info('Done');
@@ -189,7 +189,7 @@ export class InputsService {
getColumns: async (data: TestConnectionGetColumnsRequest) => {
this.logger.info('InputClientService - TestConnection/Get-Columns');
const getColumnsResponse = await new Promise((resolve, reject) => {
this.inputService.GetColumns(objectSnakeToCamel(data)).subscribe({
this.inputReadService.GetColumns(data).subscribe({
next(x) {
resolve(objectCamelToSnake(x));
},
@@ -214,9 +214,9 @@ export class InputsService {
},
getAvailableEntities: async (data) => {
this.logger.info('InputClientService - GetAvailableEntities');
objectSnakeToCamel(data);
data;
const response = await lastValueFrom(
this.inputService.GetAvailableEntities(data),
this.inputReadService.GetAvailableEntities(data),
).catch((err) => {
this.logger.info(err.details);
if (err.details && err.details.includes('400'))
@@ -228,28 +228,22 @@ export class InputsService {
return response;
},
};
constructor(
@Inject(DadosferaLogger)
dadosferaLogger: DadosferaLogger,
@Inject(InputsGrpcClient.name) private readonly grpcClient: ClientGrpc,
) {
this.logger = dadosferaLogger.logger;
}
async onModuleInit() {
this.inputService = this.grpcClient.getService<InputService>(
InputServicesNames.InputService,
);
}
secondsInADay = 60 * 60 * 24;
secondsInAnHour = 60 * 60;
adjustInputPayload(payload) {
if (payload?.input_generic) return DecodeGrpcStruct(payload.input_generic);
if (!payload.input_s3 && !payload.input_jdbc) return payload;
return payload?.input_s3 || payload?.input_jdbc;
if (payload) {
if (payload.input_generic) {
const { credentials } = payload.input_generic;
return {
...payload.input_generic,
credentials: credentials && JSON.parse(credentials),
};
}
if (payload.input_s3) return payload.input_s3;
if (payload.input_jdbc) return payload.input_jdbc;
}
return payload;
}
getDifferenceInSeconds(date1: Date, date2: Date) {
const diffInMs = Math.abs(date2.getTime() - date1.getTime());
@@ -319,20 +313,39 @@ export class InputsService {
async getAvailableEntities(data): Promise<{ entities: string[] }> {
return await this.OLD_inputClient.getAvailableEntities(data);
}
async findAll(body) {
try {
const findAllInputResponse: any = await this.OLD_inputClient.findAll(
body,
);
if (findAllInputResponse?.inputs?.length) {
findAllInputResponse.inputs = findAllInputResponse.inputs.map((input) =>
this.adjustInputPayload(input),
);
}
return findAllInputResponse;
} catch (err) {
throw new HttpException(err.message, HttpStatus.NOT_FOUND);
}
async findAll(info: Info) {
const { inputs } = await lastValueFrom(
this.inputReadService.InputFindAll({ info }),
);
if (!inputs) return { inputs: [] };
const inputsReturn = inputs.map((i) => this.adjustInputPayload(i));
return { inputs: inputsReturn };
// findAll: async (data) => {
// this.logger.info('InputClientService - FindAll');
// const findAllInputResponse = await lastValueFrom(
// this.inputReadService.InputFindAll(data),
// ).catch((e) => {
// this.logger.error(e.details);
// throw new HttpException(e.details, HttpStatus.INTERNAL_SERVER_ERROR);
// });
// findAllInputResponse.inputs.forEach((input) => objectCamelToSnake(input));
// return findAllInputResponse;
// },
// try {
// const findAllInputResponse: any = await this.OLD_inputClient.findAll(
// body,
// );
// if (findAllInputResponse?.inputs?.length) {
// findAllInputResponse.inputs = findAllInputResponse.inputs.map((input) =>
// this.adjustInputPayload(input),
// );
// }
// return findAllInputResponse;
// } catch (err) {
// throw new HttpException(err.message, HttpStatus.NOT_FOUND);
// }
}
async findOne(idRequest: IIdRequest) {