mirror of
https://github.com/dadosfera/maestro.git
synced 2026-11-02 07:19:08 +00:00
Merge pull request #208 from dadosfera/fix/upload-files
FIX: create pipeline from any file type on upload
This commit is contained in:
@@ -10,6 +10,8 @@ export class IPipelineV2 {
|
||||
@ApiProperty()
|
||||
description: string;
|
||||
@ApiProperty()
|
||||
type: string;
|
||||
@ApiProperty()
|
||||
connection_id: string;
|
||||
@ApiProperty()
|
||||
cron: string;
|
||||
@@ -104,4 +106,20 @@ export class ICreatePipelineCSVFile {
|
||||
|
||||
@ApiProperty()
|
||||
header: boolean;
|
||||
|
||||
@ApiProperty()
|
||||
engine: string;
|
||||
}
|
||||
|
||||
export class PipelineFindAllReq {
|
||||
@ApiPropertyOptional()
|
||||
search?: string | undefined;
|
||||
@ApiPropertyOptional()
|
||||
filters?: string | undefined;
|
||||
@ApiPropertyOptional()
|
||||
size?: string | undefined;
|
||||
@ApiPropertyOptional()
|
||||
page?: string | undefined;
|
||||
@ApiPropertyOptional()
|
||||
type?: string | undefined;
|
||||
}
|
||||
|
||||
@@ -40,6 +40,7 @@ import {
|
||||
ICreatePipelineV2Req,
|
||||
IPipelineV2,
|
||||
IInitUploadCSVFile,
|
||||
PipelineFindAllReq,
|
||||
} from './interfaces';
|
||||
import { GrpcToHttpExceptionFilter } from 'src/error/grpc-to-http-exception.filter';
|
||||
import { LanguageEnum } from 'src/utils/languages.enum';
|
||||
@@ -115,7 +116,7 @@ export class PipelinesController {
|
||||
async findAll(
|
||||
@User() user: RequestUser,
|
||||
@Language() language: LanguageEnum,
|
||||
@Query() data,
|
||||
@Query() data: PipelineFindAllReq,
|
||||
) {
|
||||
this.logger.info('PipelinesController - findAll', { user });
|
||||
|
||||
@@ -370,30 +371,37 @@ export class PipelinesController {
|
||||
@User() user: RequestUser,
|
||||
@Body() body: ICreatePipelineCSVFile,
|
||||
) {
|
||||
this.logger.info('/upload - Upload Connector Route');
|
||||
this.logger.info('/file - Upload Connector Route');
|
||||
const metadata = PackTheMetadata({ ...user });
|
||||
|
||||
const { name, sep, header, encoding, description, file_name } = body;
|
||||
const { name, sep, header, encoding, description, file_name, engine } =
|
||||
body;
|
||||
|
||||
const file_format_params = {
|
||||
...(sep && { sep }),
|
||||
...(encoding && { encoding }),
|
||||
...(header !== undefined ? { header } : {}),
|
||||
};
|
||||
|
||||
const source_prefix = `${user.customer_name}/${file_name}`;
|
||||
|
||||
const upload_pipeline = {
|
||||
const upload_pipeline: ICreatePipelineV2Req = {
|
||||
connection_id: process.env.UPLOAD_FILE_AGENT_CONNECTION,
|
||||
connector_name: 'Amazon S3',
|
||||
connector_plugin: 'aws_s3',
|
||||
connector_version: '1.0.0',
|
||||
image_url: 'https://assets.dadosfera.ai/images/connectors/csv.svg',
|
||||
image_url: `https://assets.dadosfera.ai/images/connectors/${engine}.svg`,
|
||||
name,
|
||||
description,
|
||||
transformations_ids: [],
|
||||
tags: [],
|
||||
cron: '@once',
|
||||
config: { cron: '@once', tables: [] },
|
||||
type: 'upload',
|
||||
properties: {
|
||||
engine: 'csv',
|
||||
engine,
|
||||
source_bucket: process.env.BUCKET_CUSTOMER_CSV_ASSETS,
|
||||
source_prefix,
|
||||
file_format_params: { sep, encoding, header: Boolean(header) },
|
||||
is_a_upload_csv: true,
|
||||
file_format_params,
|
||||
},
|
||||
input_id: undefined,
|
||||
};
|
||||
|
||||
@@ -23,7 +23,7 @@ import { ConnectorClientService } from '../connector/client.service';
|
||||
import { InputsService } from '../inputs/inputs.service';
|
||||
import { RequestUser } from 'src/decorators/user.decorator';
|
||||
import { TransformationsService } from '../transformations/transformations.service';
|
||||
import { getObjValueFromPath } from 'src/utils/ObjValueFromPath';
|
||||
import { getObjValueFromPath, objHasPath } from 'src/utils/ObjValueFromPath';
|
||||
import ErrorCodes from 'src/utils/errorCodes';
|
||||
import ErrorBuilder from 'src/utils/ErrorBuilder';
|
||||
|
||||
@@ -220,7 +220,7 @@ export class PipelinesService implements OnModuleInit {
|
||||
this.logger.info('PipelinesClientService - findOneProperties');
|
||||
const { pipeline } = await this.findOne({ id }, metadata);
|
||||
|
||||
const pipelineProperties = JSON.parse(pipeline.properties);
|
||||
const properties = JSON.parse(pipeline.properties);
|
||||
|
||||
const connectorId = `${pipeline.connector_plugin}-${pipeline.connector_version}`;
|
||||
|
||||
@@ -228,24 +228,30 @@ export class PipelinesService implements OnModuleInit {
|
||||
connectorId,
|
||||
metadata.get('language')[0].toString(),
|
||||
);
|
||||
const connector = JSON.parse(res.connector);
|
||||
const config_controls = connector.config_controls || [];
|
||||
const connector: { [key: string]: any; config_controls: any[] } =
|
||||
JSON.parse(res.connector);
|
||||
const config_controls = (connector.config_controls || []).filter(
|
||||
(config_control) => {
|
||||
const path: string = config_control.name;
|
||||
|
||||
for (const config_control of config_controls) {
|
||||
const path: string = config_control.name;
|
||||
const default_value = getObjValueFromPath({
|
||||
object: { properties: pipelineProperties },
|
||||
path,
|
||||
});
|
||||
if (default_value != undefined)
|
||||
config_control.default_value = default_value;
|
||||
}
|
||||
if (!objHasPath({ object: { properties }, path })) return false;
|
||||
|
||||
const default_value = getObjValueFromPath({
|
||||
object: { properties },
|
||||
path,
|
||||
});
|
||||
if (default_value != undefined)
|
||||
config_control.default_value = default_value;
|
||||
|
||||
return true;
|
||||
},
|
||||
);
|
||||
|
||||
return {
|
||||
pipeline: {
|
||||
...pipeline,
|
||||
config_controls,
|
||||
properties: pipelineProperties,
|
||||
properties: properties,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -308,6 +308,20 @@ export function EnrichErrorCode(code: string) {
|
||||
'Tente realizar a ação novamente. Caso o erro persista, entre em contato com o suporte',
|
||||
code,
|
||||
};
|
||||
case ErrorCodes.CATALOG.PREVIEW_TOO_BIG:
|
||||
return {
|
||||
statusCode: HttpStatus.INTERNAL_SERVER_ERROR,
|
||||
error: 'Erro ao buscar prévia',
|
||||
message: 'Ocorreu um erro devido ao tamanho da prévia do ativo.',
|
||||
code,
|
||||
};
|
||||
case ErrorCodes.CATALOG.METADATA_TOO_BIG:
|
||||
return {
|
||||
statusCode: HttpStatus.INTERNAL_SERVER_ERROR,
|
||||
error: 'Erro ao buscar informações',
|
||||
message: 'Ocorreu um erro devido ao tamanho das informações do ativo.',
|
||||
code,
|
||||
};
|
||||
case ErrorCodes.INTERNAL:
|
||||
case ErrorCodes.UNKNOWN:
|
||||
default:
|
||||
|
||||
@@ -8,3 +8,18 @@ export function getObjValueFromPath<T>(data: { object: T; path: string }): T {
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
export function objHasPath<T extends object>(data: {
|
||||
object: T;
|
||||
path: string;
|
||||
}): boolean {
|
||||
const { object, path } = data;
|
||||
const keys = path.split('.');
|
||||
|
||||
let value = JSON.parse(JSON.stringify(object));
|
||||
for (const key of keys) {
|
||||
if (!(key in value)) return false;
|
||||
value = value[key];
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -70,6 +70,8 @@ const TRANSFORMATION = {
|
||||
};
|
||||
const CATALOG = {
|
||||
DATA_ASSET_NOT_FOUND: 'CATALOG.DATA_ASSET_NOT_FOUND',
|
||||
PREVIEW_TOO_BIG: 'CATALOG.PREVIEW_TOO_BIG',
|
||||
METADATA_TOO_BIG: 'CATALOG.METADATA_TOO_BIG',
|
||||
};
|
||||
const ErrorCodes = {
|
||||
UNKNOWN: 'UNKNOWN',
|
||||
|
||||
Reference in New Issue
Block a user