Compare commits

...
21 Commits
Author SHA1 Message Date
Gabriel Amorim 4df23c917a FEAT: new connectors
Merge pull request #103 from dadosfera/feature/hubspot-connector
2022-06-21 11:44:08 -03:00
Gabriel Rosa 6c3eef7181 update protospack and deploy file 2022-06-21 10:13:20 -03:00
Gabriel Rosa 668c5fe6a5 WIP: Documentation of input API 2022-06-20 11:36:26 -03:00
Gabriel Rosa d057495b2e FIX: redirect_uri.redirect_uri fixed 2022-06-15 15:19:45 -03:00
Gabriel Rosa 2f1c71eefe update protospack and better adjustInputPayload 2022-06-15 11:14:22 -03:00
Gabriel Rosa 3bb014e714 get redirect_uri from get_tokens_url_params 2022-06-15 10:08:03 -03:00
Gabriel Rosa 2ff2aa049f update protospack 2022-06-15 08:52:30 -03:00
Gabriel Rosa f6c7420f59 finding inputs using generic strategy 2022-06-15 08:44:21 -03:00
Gabriel Rosa abbe88f45b removed unused statement 2022-06-14 14:02:06 -03:00
Gabriel Rosa 383741f48c get entities 2022-06-14 14:01:40 -03:00
Gabriel Rosa ff663e7166 update protospack 2022-06-14 11:35:09 -03:00
Gabriel Rosa 305daeeeab update protospack 2022-06-13 16:50:38 -03:00
Gabriel Rosa 852ddbf88b transforming to camelCase before sending to in-factory 2022-06-13 15:55:02 -03:00
Gabriel Rosa 4dfe6c3870 fallback for env vars 2022-06-13 13:12:01 -03:00
Gabriel Rosa 2710159a73 minor improvements 2022-06-13 11:17:41 -03:00
Gabriel Rosa 1ab5d49908 removing info that won't be saved on database 2022-06-13 10:49:27 -03:00
Gabriel Amorim 43372b5a0f FIX: Better error handling on input creation
Merge pull request #102 from dadosfera/fix/error-handling
2022-06-10 16:05:29 -03:00
Gabriel Rosa 0ac2dd2de7 Better error handling on input creation 2022-06-10 15:57:19 -03:00
Gabriel Rosa 0342386a76 oauth almost complete 2022-06-09 17:14:11 -03:00
Gabriel Rosa 28e55f8fd9 update node version 2022-06-08 11:12:51 -03:00
Gabriel Rosa 4e2e296442 Fixed cron on create S3 2022-06-08 10:31:59 -03:00
13 changed files with 436 additions and 122 deletions
+3 -3
View File
@@ -134,7 +134,7 @@ jobs:
eb deploy
- name: Remove Docker's Trash
continue-on-error: true
if: always()
run: |
docker system prune
docker rmi -f $(docker images -aq)
docker system prune --volumes -a -f
docker system df
+1 -1
View File
@@ -1 +1 @@
18
18.3.0
+2 -1
View File
@@ -1,7 +1,8 @@
FROM node:14.15.4-alpine3.12 as packages
FROM node:18.3.0-alpine3.15 as packages
WORKDIR /packages
ARG NPM_TOKEN
COPY package.json .
COPY package-lock.json .
COPY .npmrc .
RUN npm install
RUN rm -f ./.npmrc
+10 -7
View File
@@ -19,7 +19,7 @@
"@nestjs/platform-express": "^8.4.3",
"@nestjs/schedule": "^1.1.0",
"@nestjs/swagger": "^5.2.1",
"@victorradael/protospack": "2.4.2-1",
"@victorradael/protospack": "2.5.0",
"axios": "^0.25.0",
"cron-parser": "^4.4.0",
"dotenv": "^14.2.0",
@@ -56,6 +56,9 @@
"ts-node": "^10.7.0",
"tsconfig-paths": "^3.14.1",
"typescript": "^4.6.3"
},
"engines": {
"node": "18.3.0"
}
},
"node_modules/@ampproject/remapping": {
@@ -2357,9 +2360,9 @@
}
},
"node_modules/@victorradael/protospack": {
"version": "2.4.2-1",
"resolved": "https://registry.npmjs.org/@victorradael/protospack/-/protospack-2.4.2-1.tgz",
"integrity": "sha512-wdcGaK0DIEJeOHga1br11kq264JNdhdB1G2toy/u6TvAk90wwBKHNs/cZko0p9WFVuqHQOVehoHj7WJbJ3K25Q==",
"version": "2.5.0",
"resolved": "https://registry.npmjs.org/@victorradael/protospack/-/protospack-2.5.0.tgz",
"integrity": "sha512-DORlm3uHAYVH9dlcmwyOxjA4rk1UZWFyYUpD02k0bFfrH+uzSWe8qiPtTgZym9P8U9D8bXYRWeLlroasBj2Rdw==",
"dependencies": {
"rxjs": "^7.5.5"
}
@@ -10932,9 +10935,9 @@
}
},
"@victorradael/protospack": {
"version": "2.4.2-1",
"resolved": "https://registry.npmjs.org/@victorradael/protospack/-/protospack-2.4.2-1.tgz",
"integrity": "sha512-wdcGaK0DIEJeOHga1br11kq264JNdhdB1G2toy/u6TvAk90wwBKHNs/cZko0p9WFVuqHQOVehoHj7WJbJ3K25Q==",
"version": "2.5.0",
"resolved": "https://registry.npmjs.org/@victorradael/protospack/-/protospack-2.5.0.tgz",
"integrity": "sha512-DORlm3uHAYVH9dlcmwyOxjA4rk1UZWFyYUpD02k0bFfrH+uzSWe8qiPtTgZym9P8U9D8bXYRWeLlroasBj2Rdw==",
"requires": {
"rxjs": "^7.5.5"
}
+4 -1
View File
@@ -5,6 +5,9 @@
"author": "",
"private": true,
"license": "UNLICENSED",
"engines": {
"node": "18.3.0"
},
"scripts": {
"prebuild": "rimraf dist",
"build": "nest build",
@@ -31,7 +34,7 @@
"@nestjs/platform-express": "^8.4.3",
"@nestjs/schedule": "^1.1.0",
"@nestjs/swagger": "^5.2.1",
"@victorradael/protospack": "2.4.2-1",
"@victorradael/protospack": "2.5.0",
"axios": "^0.25.0",
"cron-parser": "^4.4.0",
"dotenv": "^14.2.0",
+135 -81
View File
@@ -1,17 +1,27 @@
import { OnModuleInit, Inject } from '@nestjs/common';
import {
OnModuleInit,
Inject,
HttpException,
HttpStatus,
} from '@nestjs/common';
import { ClientGrpc } from '@nestjs/microservices';
import {
InputCreateResponse,
DecodeGrpcStruct,
EncodeJsonToGrpcStruct,
InputCreateS3Request,
InputNewCreateRequest,
InputService,
TestConnectionGetColumnsRequest,
TestConnectionRequest,
} from '@victorradael/protospack';
import axios from 'axios';
import { lastValueFrom } from 'rxjs';
import { InputModel } from 'src/modules/inputs/dtos/input.model';
import {
objectCamelToSnake,
objectSnakeToCamel,
} from 'src/utils/CaseConverter';
import { mustache } from 'src/utils/mustache';
import { IIdRequest, UpdateInputRequest } from './interfaces';
export class InputsClientService implements OnModuleInit {
@@ -25,102 +35,68 @@ export class InputsClientService implements OnModuleInit {
this.grpcClient.getService<InputService>('InputService');
}
async create(createInputDto: InputNewCreateRequest) {
async newCreate(createInputDto: InputNewCreateRequest) {
console.log('InputClientService', 'Create');
const createInputResponse = await new Promise((resolve, reject) => {
this.inputService
.NewCreate(objectSnakeToCamel(createInputDto))
.subscribe({
next(x) {
resolve(objectCamelToSnake(x));
},
error(err) {
console.log('Observable Error');
reject(err);
},
complete() {
console.log('done');
},
});
}).catch((err) => {
throw new Error(err);
});
const input = await lastValueFrom(
this.inputService.NewCreate(objectSnakeToCamel(createInputDto)),
);
objectCamelToSnake(input);
return input;
}
return createInputResponse;
async createGeneric(createInputGeneric) {
console.log('InputClientService', 'Create');
const { info, ...data } = createInputGeneric;
data.credentials = await this.getInputTokens(data);
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),
);
objectCamelToSnake(structReturn);
const inputCreated = DecodeGrpcStruct(structReturn.input);
return { input: inputCreated };
}
async createS3Inputs(createInputDto: InputCreateS3Request) {
console.log('InputClientService', 'Create');
const createInputResponse = await new Promise((resolve, reject) => {
this.inputService.CreateS3(objectSnakeToCamel(createInputDto)).subscribe({
next(x) {
resolve(objectCamelToSnake(x));
},
error(err) {
console.log('Observable Error');
reject(err);
},
complete() {
console.log('done');
},
});
})
.then((res) => res)
.catch((err) => {
throw new Error(err);
});
objectSnakeToCamel(createInputDto);
const createInputResponse = await lastValueFrom(
this.inputService.CreateS3(createInputDto),
).catch((e) => {
throw new HttpException(e.details, 500);
});
console.log('done');
objectCamelToSnake(createInputResponse);
return createInputResponse;
}
async findOne(data: IIdRequest) {
console.log('InputClientService', 'FindOne');
const findOneInputResponse = await new Promise((resolve, reject) => {
this.inputService.FindOne(objectSnakeToCamel(data)).subscribe({
next(x) {
resolve(objectCamelToSnake(x));
},
error(err) {
console.log('Observable Error');
reject(err);
},
complete() {
console.log('done');
},
});
})
.then((res) => res)
.catch((err) => {
throw new Error(err);
});
const findOneInputResponse = await lastValueFrom(
this.inputService.FindOne(objectSnakeToCamel(data)),
).catch((e) => {
throw new HttpException(e.details, HttpStatus.INTERNAL_SERVER_ERROR);
});
objectCamelToSnake(findOneInputResponse);
return findOneInputResponse;
}
async findAll(data) {
console.log('InputClientService', 'FindAll');
const findAllInputResponse = await new Promise((resolve, reject) => {
this.inputService.FindAll(objectSnakeToCamel(data)).subscribe({
next(x) {
resolve(objectCamelToSnake(x));
},
error(err) {
console.log('Observable Error');
reject(err);
},
complete() {
console.log('done');
},
});
})
.then((res) => res)
.catch((err) => {
throw new Error(err);
});
const findAllInputResponse = await lastValueFrom(
this.inputService.FindAll(objectSnakeToCamel(data)),
).catch((e) => {
throw new HttpException(e.details, HttpStatus.INTERNAL_SERVER_ERROR);
});
findAllInputResponse.inputs.forEach((input) => objectCamelToSnake(input));
return findAllInputResponse;
}
@@ -221,4 +197,82 @@ export class InputsClientService implements OnModuleInit {
return getColumnsResponse;
}
async getAuthSecrets(plugin) {
switch (plugin) {
case 'hubspot':
return {
client_id: process.env.HUBSPOT_CLIENT_ID || '',
client_secret: process.env.HUBSPOT_CLIENT_SECRET || '',
};
}
return {};
}
async getInputTokens(input: InputModel) {
const { credentials, options } = input;
const credentialsTokens: any = {};
switch (credentials.connection_type) {
case 'oauth':
const { oauth } = options;
const { get_tokens_url_params, get_tokens_set_response } = oauth;
const secrets = await this.getAuthSecrets(input.plugin);
if (oauth.content_type === 'application/x-www-form-urlencoded') {
const requestParams = new URLSearchParams();
const get_tokens_url_params_string = mustache(get_tokens_url_params, {
...secrets,
...credentials,
});
let get_tokens_url_params_obj: Record<string, any> = {};
try {
get_tokens_url_params_obj = JSON.parse(
get_tokens_url_params_string,
);
} catch (error) {
throw new HttpException(
'Erro transformando get_tokens_url_params',
400,
);
}
const { redirect_uri } = get_tokens_url_params_obj;
for (const key in get_tokens_url_params_obj) {
const value = get_tokens_url_params_obj[key];
requestParams.append(key, value);
}
const { data } = await axios
.post(options.oauth.get_tokens_url, requestParams)
.catch((err) => {
console.log(err.response.data);
throw new HttpException(
err.response.data.message,
err.response.status,
);
});
for (const key in get_tokens_set_response) {
const responseKey = get_tokens_set_response[key];
credentialsTokens[key] = data[responseKey];
}
credentialsTokens.redirect_uri = redirect_uri;
}
break;
}
return { ...credentials, ...credentialsTokens };
}
// getOauthTokens(input){
// }
async getAvailableEntities(data) {
console.log('InputClientService', 'GetAvailableEntities');
objectSnakeToCamel(data);
const response = await lastValueFrom(
this.inputService.GetAvailableEntities(data),
).catch((err) => {
console.log(err);
if (err.details && err.details.includes('400'))
throw new HttpException('Plugin inválido', HttpStatus.BAD_REQUEST);
throw new HttpException(err.details, HttpStatus.INTERNAL_SERVER_ERROR);
});
objectCamelToSnake(response);
return response;
}
}
+5 -2
View File
@@ -1,4 +1,4 @@
import { Inject, OnModuleInit } from '@nestjs/common';
import { HttpException, Inject, OnModuleInit } from '@nestjs/common';
import { ClientGrpc, Payload } from '@nestjs/microservices';
import {
PipelineServicesNames,
@@ -49,7 +49,10 @@ export class PipelinesClientService implements OnModuleInit {
})
.then((res) => res)
.catch((err) => {
throw new Error(err);
if (err.details == 'Invalid Request') {
throw new HttpException(err.details, 400);
}
throw new HttpException(err.details, 500);
});
return createPipelineResponse;
+202
View File
@@ -0,0 +1,202 @@
import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
export class InputModel {
category: string;
plugin: string;
name: string;
cron?: string;
credentials?: {
connection_type: string;
oauth: string;
oauth_code: string;
start_date: string;
};
options?: {
oauth?: {
get_tokens_url: string;
get_tokens_url_params: string;
get_tokens_set_response: Record<string, string>;
content_type: string;
};
};
}
class GoogleAnalyticsClientSecrets {
@ApiProperty()
type: string;
@ApiProperty()
project_id: string;
@ApiProperty()
private_key_id: string;
@ApiProperty()
private_key: string;
@ApiProperty()
client_email: string;
@ApiProperty()
client_id: string;
@ApiProperty()
auth_uri: string;
@ApiProperty()
token_uri: string;
@ApiProperty()
auth_provider_x509_cert_url: string;
@ApiProperty()
client_x509_cert_url: string;
}
class OauthObject {
@ApiPropertyOptional()
get_tokens_url: string;
@ApiPropertyOptional()
get_tokens_url_params: string;
@ApiPropertyOptional()
get_tokens_set_response: Record<string, any>;
@ApiPropertyOptional()
content_type: string;
}
class InputOptions {
@ApiPropertyOptional()
oauth: OauthObject;
@ApiPropertyOptional()
skip_select_columns: true;
@ApiPropertyOptional()
skip_select_entities: false;
@ApiPropertyOptional()
skip_transformation: true;
}
class CredentialsJdbc {
@ApiPropertyOptional()
jdbc_user: string;
@ApiPropertyOptional()
jdbc_password: string;
@ApiPropertyOptional()
database: string;
@ApiPropertyOptional()
endpoint: string;
@ApiPropertyOptional()
port: string;
@ApiPropertyOptional()
engine: string;
@ApiPropertyOptional()
schema: string;
}
class FileFormatParams {
@ApiPropertyOptional()
file_format: string;
@ApiPropertyOptional()
encoding: string;
}
class Credentials extends CredentialsJdbc {
@ApiProperty()
connection_type: string;
@ApiPropertyOptional()
client_aws_access_key_id: string;
@ApiPropertyOptional()
client_aws_secret_access_key: string;
@ApiPropertyOptional()
client_bucket: string;
@ApiPropertyOptional()
file_to_extract: string;
@ApiPropertyOptional()
file_format_params: FileFormatParams;
@ApiPropertyOptional()
view_id: string;
@ApiPropertyOptional()
client_secrets: GoogleAnalyticsClientSecrets;
@ApiPropertyOptional()
start_date: string;
@ApiPropertyOptional()
end_date: string;
@ApiPropertyOptional()
oauth_code: string;
}
class Column {
@ApiProperty()
name: string;
@ApiProperty()
type: string;
}
class TableColumns {
@ApiProperty()
name: string;
@ApiProperty()
columns: Column[];
@ApiProperty()
references: Column[];
}
export class GetAvailableEntitiesReq {
@ApiProperty()
plugin: string;
}
export class GetAvailableEntitiesRes {
@ApiProperty()
entities: string[];
}
export class TestConnectionGetColumnsReq {
@ApiProperty()
plugin: string;
@ApiProperty()
tables: string[];
@ApiPropertyOptional()
credentials: CredentialsJdbc;
@ApiPropertyOptional()
id: string;
}
export class TestConnectionGetColumnsRes {
@ApiProperty()
tables: TableColumns[];
}
export class TestConnectionReq {
@ApiProperty()
plugin: string;
@ApiProperty()
credentials: CredentialsJdbc;
}
export class TestConnectionRes {
@ApiProperty()
connection_state: boolean;
@ApiProperty()
total_entities: number;
@ApiProperty()
database_tables: string[];
}
export class CreateInputReq {
@ApiProperty()
plugin: string;
@ApiProperty()
category: string;
@ApiProperty()
credentials: Credentials;
@ApiProperty()
name: string;
@ApiPropertyOptional()
options: InputOptions;
}
export class Input {
@ApiProperty()
id: string;
@ApiProperty()
category: string;
@ApiProperty()
plugin: string;
@ApiProperty()
name: string;
@ApiProperty()
cron: string;
@ApiProperty()
credentials: Credentials;
@ApiProperty()
client_id: string;
@ApiProperty()
created_at: string;
@ApiProperty()
updated_at: string;
}
+33 -5
View File
@@ -10,14 +10,40 @@ import {
import { InputsService } from './inputs.service';
import { InputsClientService } from 'src/clients/inputs/client.service';
import { UpdateInputRequest } from 'src/clients/inputs/interfaces';
import { InputNewCreateRequest } from '@victorradael/protospack';
import { ApiOkResponse, ApiTags } from '@nestjs/swagger';
import {
CreateInputReq,
GetAvailableEntitiesReq,
GetAvailableEntitiesRes,
Input,
TestConnectionGetColumnsReq,
TestConnectionGetColumnsRes,
TestConnectionReq,
TestConnectionRes,
} from './dtos/input.model';
@ApiTags('inputs')
@Controller('inputs')
export class InputsController {
constructor(private inputsClientService: InputsClientService) {}
inputService: InputsService;
constructor(private inputsClientService: InputsClientService) {
this.inputService = new InputsService(this.inputsClientService);
}
@Get('available-entities/:plugin')
@ApiOkResponse({ type: GetAvailableEntitiesRes })
async getAvailableEntities(
@Body() body,
@Param() params: GetAvailableEntitiesReq,
) {
const { info } = body;
const { plugin } = params;
return await this.inputService.getAvailableEntities({ info, plugin });
}
@Post('/test-connection')
async testConnection(@Body() data) {
@ApiOkResponse({ type: TestConnectionRes })
async testConnection(@Body() data: TestConnectionReq) {
console.log(`/test-connection`, 'ON TEST CONNECTION ROUTE');
const inputService = new InputsService(this.inputsClientService);
@@ -27,7 +53,8 @@ export class InputsController {
}
@Post('/test-connection/get-columns')
async getColumns(@Body() data) {
@ApiOkResponse({ type: TestConnectionGetColumnsRes })
async getColumns(@Body() data: TestConnectionGetColumnsReq) {
console.log(
`/test-connection/get-columns`,
'ON TEST CONNECTION GET COLUMNS ROUTE',
@@ -40,7 +67,8 @@ export class InputsController {
}
@Post()
async create(@Body() createInputDto: InputNewCreateRequest) {
@ApiOkResponse({ type: Input })
async create(@Body() createInputDto: CreateInputReq) {
console.log(`/input`, 'ON CREATE ROUTE');
const inputService = new InputsService(this.inputsClientService);
+27 -19
View File
@@ -1,5 +1,6 @@
import { Body, HttpException, HttpStatus, Injectable } from '@nestjs/common';
import { Timeout } from '@nestjs/schedule';
import { DecodeGrpcStruct } from '@victorradael/protospack';
import CronParser, { CronExpression } from 'cron-parser';
import { InputsClientService } from 'src/clients/inputs/client.service';
import { IIdRequest, Info } from 'src/clients/inputs/interfaces';
@@ -11,6 +12,8 @@ export class InputsService {
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;
}
getDifferenceInSeconds(date1: Date, date2: Date) {
@@ -47,31 +50,34 @@ export class InputsService {
}
async create(@Body() data) {
this.validateCron(data);
try {
if (
data.plugin == 'csv' ||
data.plugin == 'json' ||
data.plugin == 'parquet'
) {
let response;
switch (data.plugin.toLowerCase()) {
case 'csv':
case 'json':
case 'parquet':
const { info, ...input } = data;
const inputPayload = this.generateInputS3Payload(input);
const createInputResponse = await this.inputClient.createS3Inputs({
response = await this.inputClient.createS3Inputs({
input: inputPayload,
info,
});
return createInputResponse;
} else {
const createInputResponse: any = await this.inputClient.create(data);
if (createInputResponse?.input?.input_jdbc) {
return { input: createInputResponse?.input?.input_jdbc };
}
return createInputResponse;
}
} catch (err) {
throw new HttpException(err.message, HttpStatus.NOT_FOUND);
break;
case 'oracle':
case 'mysql':
case 'postgresql':
case 'sqlserver':
response = await this.inputClient.newCreate(data);
break;
default:
response = await this.inputClient.createGeneric(data);
break;
}
const adjustedInput = this.adjustInputPayload(response.input);
return { ...response, input: adjustedInput };
}
async getAvailableEntities(data): Promise<{ entities: string[] }> {
return await this.inputClient.getAvailableEntities(data);
}
async findAll(body) {
try {
const findAllInputResponse: any = await this.inputClient.findAll(body);
@@ -153,7 +159,7 @@ export class InputsService {
}
generateInputS3Payload(payload) {
const { credentials, plugin } = payload;
const { credentials, plugin, cron } = payload;
if (!credentials) return payload;
const {
@@ -164,8 +170,10 @@ export class InputsService {
client_bucket,
file_to_extract,
} = credentials;
const formatedPayload = {
plugin,
cron,
source_bucket: client_bucket,
source_prefix: file_to_extract,
auth_parameters: {
+6 -1
View File
@@ -1,4 +1,5 @@
import { HttpException, HttpStatus, Injectable } from '@nestjs/common';
import { DecodeGrpcStruct } from '@victorradael/protospack';
import { PipelinesClientService } from 'src/clients/pipelines/client.service';
import { IIdRequest, Info } from 'src/clients/pipelines/interfaces';
import { objectCamelToSnake } from 'src/utils/CaseConverter';
@@ -8,7 +9,11 @@ export class PipelinesService {
constructor(private pipelineClient: PipelinesClientService) {}
adjustPayload(payload) {
payload.input = payload.input?.input_s3 || payload.input?.input_jdbc;
if (payload.input?.input_generic) {
payload.input = DecodeGrpcStruct(payload.input?.input_generic);
} else {
payload.input = payload.input?.input_s3 || payload.input?.input_jdbc;
}
}
async create(createPipelineDto) {
+7
View File
@@ -0,0 +1,7 @@
export const mustache = (string: string, data: Record<string, any> = {}) => {
return Object.entries(data).reduce((res, [key, value]) => {
const mainRe = new RegExp(`(?<!\\\\){{\\s*${key}\\s*}}`, 'g');
const escapeRe = new RegExp(`\\\\({{\\s*${key}\\s*}})`, 'g');
return res.replace(mainRe, value.toString()).replace(escapeRe, '$1');
}, string);
};
+1 -1
View File
File diff suppressed because one or more lines are too long