mirror of
https://github.com/dadosfera/maestro.git
synced 2026-10-17 03:19:09 +00:00
846 lines
25 KiB
TypeScript
846 lines
25 KiB
TypeScript
import DadosferaLogger from '@dadosfera/dadosfera-logs/dist';
|
|
import {
|
|
WriteService,
|
|
ReadService,
|
|
ProtoServices,
|
|
Messages,
|
|
} from '@dadosfera/protospack-v2/dist/lib/Catalog';
|
|
import {
|
|
Messages as PlatformInterfaceMessages,
|
|
WriteService as PlatformInterfaceWriteService,
|
|
ProtoServices as PlatformInterfacesProtoServices,
|
|
} from '@dadosfera/protospack-v2/dist/lib/PlatformInterfaces';
|
|
import {
|
|
BadRequestException,
|
|
HttpException,
|
|
HttpStatus,
|
|
Inject,
|
|
OnModuleInit,
|
|
} from '@nestjs/common';
|
|
import { ClientGrpc } from '@nestjs/microservices';
|
|
import axios from 'axios';
|
|
import { lastValueFrom } from 'rxjs';
|
|
import { CatalogClientConfiguration } from './catalog-client';
|
|
import { UsersService } from '../users/users.service';
|
|
import { RolesService } from '../roles/roles.service';
|
|
import { Metadata } from '@grpc/grpc-js';
|
|
import { PackTheMetadata } from 'src/utils/PackTheMetadata';
|
|
import {
|
|
AssetReporter,
|
|
BatchRemoveRlsRulesRequest,
|
|
CreateDataDocsDTO,
|
|
IUpdateCertificationStatusRequest,
|
|
IUpdateDataRequest,
|
|
TriggerCatalogReq,
|
|
} from './dtos';
|
|
import {
|
|
AddRlsRuleRequest,
|
|
GetNimbusDashboardsRequest,
|
|
GetRlsRulesRequest,
|
|
PiiMetadata,
|
|
} from '@dadosfera/protospack-v2/dist/lib/Catalog/interfaces/messages';
|
|
import { TypeParser } from 'src/utils/FileParser/parser-types';
|
|
import { ParserBuilder } from 'src/utils/FileParser/parser.builder';
|
|
|
|
class CatalogService implements OnModuleInit {
|
|
catalogReadService: ReadService.CatalogReadServices;
|
|
catalogWriteService: WriteService.CatalogWriteServices;
|
|
platformWriteService: PlatformInterfaceWriteService.PlatformInterfacesWriteServices;
|
|
logger: any;
|
|
constructor(
|
|
@Inject(DadosferaLogger)
|
|
dadosferaLogger: DadosferaLogger,
|
|
@Inject(CatalogClientConfiguration.name)
|
|
private readonly grpcClient: ClientGrpc,
|
|
private readonly userService: UsersService,
|
|
private readonly roleService: RolesService,
|
|
) {
|
|
this.logger = dadosferaLogger.logger;
|
|
}
|
|
|
|
onModuleInit() {
|
|
this.catalogReadService =
|
|
this.grpcClient.getService<ReadService.CatalogReadServices>(
|
|
ProtoServices.CatalogReadServices,
|
|
);
|
|
this.catalogWriteService =
|
|
this.grpcClient.getService<WriteService.CatalogWriteServices>(
|
|
ProtoServices.CatalogWriteServices,
|
|
);
|
|
this.platformWriteService =
|
|
this.grpcClient.getService<PlatformInterfaceWriteService.PlatformInterfacesWriteServices>(
|
|
PlatformInterfacesProtoServices.PlatformInterfacesWriteServices,
|
|
);
|
|
}
|
|
|
|
_getNimbusUrl(body) {
|
|
this.logger.debug(`Body: ${JSON.stringify(body)}`);
|
|
const customer = body.info.customer.toLowerCase();
|
|
|
|
if (process.env.ENV === 'prd') {
|
|
return `https://nimbus-${customer}.dadosfera.ai`;
|
|
}
|
|
|
|
return `https://nimbus-${customer}.${process.env.ENV.replace(
|
|
'local',
|
|
'stg',
|
|
)}.dadosfera.ai`;
|
|
}
|
|
|
|
async getPiiReporter(metadata: Metadata, type: TypeParser) {
|
|
this.logger.info('getPiiReporter: ' + type);
|
|
try {
|
|
const { data } = await lastValueFrom(
|
|
this.catalogWriteService.GetPiiReporter({}, metadata),
|
|
);
|
|
this.logger.info('Finish grpc call');
|
|
|
|
const parser = ParserBuilder.build<PiiMetadata>(type);
|
|
|
|
this.logger.info('parser file to: ' + type);
|
|
const file = await parser.parse(data);
|
|
this.logger.info('finish parser');
|
|
const mimeTypes: Record<TypeParser, string> = {
|
|
csv: 'text/csv',
|
|
html: 'text/html',
|
|
pdf: 'application/pdf',
|
|
};
|
|
|
|
const timestamp = new Date().toISOString().replace(/[:.]/g, '-');
|
|
const filename = `relatorio-pii-${timestamp}.${type}`;
|
|
|
|
return {
|
|
file,
|
|
filename: filename,
|
|
type: mimeTypes[type],
|
|
};
|
|
} catch (error) {
|
|
this.logger.error(error.message);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async getCustomPropertyDefinitions(metadata: Metadata) {
|
|
return lastValueFrom(this.catalogReadService.GetCustomPropertyDefinitions({}, metadata));
|
|
}
|
|
|
|
async createDataAsset(data: Messages.CreateDataAssetRequest, metadata) {
|
|
this.logger.info('CatalogService - Manage Data assets permissions');
|
|
if (!data.embed) data.embed = undefined;
|
|
|
|
return lastValueFrom(
|
|
this.catalogWriteService.CreateDataAsset(data, metadata),
|
|
);
|
|
}
|
|
|
|
async managePermissions(data: Messages.ManagePermissionRequest, metadata) {
|
|
this.logger.info('CatalogService - Manage Data assets permissions');
|
|
|
|
return lastValueFrom(
|
|
this.catalogWriteService.ManagePermission(data, metadata),
|
|
).catch((err) => {
|
|
throw new HttpException(
|
|
err.details,
|
|
err.code === 6 ? HttpStatus.CONFLICT : 404,
|
|
);
|
|
});
|
|
}
|
|
|
|
async revokePermissions(data: Messages.RevokePermissionRequest, metadata) {
|
|
this.logger.info('CatalogService - Manage Data assets permissions');
|
|
|
|
return lastValueFrom(
|
|
this.catalogWriteService.RevokePermission(data, metadata),
|
|
).catch((err) => {
|
|
throw new HttpException(
|
|
err.details,
|
|
err.code === 6 ? HttpStatus.CONFLICT : 404,
|
|
);
|
|
});
|
|
}
|
|
|
|
async commentOnDataAsset(data: Messages.MakeACommentRequest, metadata) {
|
|
this.logger.info('CatalogService - Manage Data assets permissions');
|
|
|
|
return lastValueFrom(
|
|
this.catalogWriteService.MakeAComment(data, metadata),
|
|
).catch((err) => {
|
|
throw new HttpException(
|
|
err.details,
|
|
err.code === 6 ? HttpStatus.CONFLICT : 404,
|
|
);
|
|
});
|
|
}
|
|
|
|
async deleteComment(data: Messages.UpdateACommentRequest, metadata) {
|
|
this.logger.info('CatalogService - Manage Data assets permissions');
|
|
|
|
return lastValueFrom(
|
|
this.catalogWriteService.UpdateAComment(data, metadata),
|
|
).catch((err) => {
|
|
throw new HttpException(
|
|
err.details,
|
|
err.code === 6 ? HttpStatus.CONFLICT : 404,
|
|
);
|
|
});
|
|
}
|
|
|
|
async deleteDataAsset(data: Messages.DeleteDataAssetRequest, metadata) {
|
|
this.logger.info('CatalogService - Manage Data assets permissions');
|
|
|
|
return lastValueFrom(
|
|
this.catalogWriteService.DeleteDataAsset(data, metadata),
|
|
).catch((err) => {
|
|
throw new HttpException(
|
|
err.details,
|
|
err.code === 6 ? HttpStatus.CONFLICT : 404,
|
|
);
|
|
});
|
|
}
|
|
|
|
async getUserRolesIds(userId: string) {
|
|
const result = await this.userService.findOneById(userId).catch(() => null);
|
|
|
|
if (result) {
|
|
return result.user.roles.map((role) => role.id);
|
|
}
|
|
|
|
return [];
|
|
}
|
|
|
|
async searchDataAssets(
|
|
query: Record<string, any>,
|
|
metadata: Metadata,
|
|
customer_id: string,
|
|
) {
|
|
this.logger.info('CatalogService - searchDataAssets', { query });
|
|
|
|
const { search, page, size, sort_by, order, ...filters } = query;
|
|
|
|
this.logger.debug('Extracted filters:', { filters });
|
|
|
|
if (
|
|
filters.manually !== undefined &&
|
|
filters.manually !== null &&
|
|
filters.manually !== ''
|
|
) {
|
|
filters.manually = Number(filters.manually); // 1 ou 0
|
|
} else {
|
|
delete filters.manually;
|
|
}
|
|
|
|
if (filters.owner) {
|
|
const { users: customer_users } =
|
|
await this.userService.findAllUsersByCustomerId(customer_id);
|
|
|
|
this.logger.info('Available users in database count:', {
|
|
count: customer_users.length,
|
|
});
|
|
this.logger.info('First 5 users:', {
|
|
users: customer_users
|
|
.slice(0, 5)
|
|
.map((u) => ({ id: u.id, email: u.email, name: u.name })),
|
|
});
|
|
|
|
const ownerValues = Array.isArray(filters.owner)
|
|
? filters.owner
|
|
: typeof filters.owner === 'string' && filters.owner.includes(',')
|
|
? filters.owner.split(',').map((o: string) => o.trim())
|
|
: [filters.owner];
|
|
|
|
this.logger.info('Owner values to convert:', {
|
|
ownerValues,
|
|
ownerFiltersOriginal: filters.owner,
|
|
});
|
|
|
|
const ownerIds = ownerValues
|
|
.map((ownerValue: string) => {
|
|
const normalizedOwner = ownerValue.replace(/\s/g, '+');
|
|
const user = customer_users.find((u) => {
|
|
const isIdMatch = u.id === ownerValue;
|
|
const isEmailMatch =
|
|
u.email === ownerValue || u.email === normalizedOwner;
|
|
const isNameMatch =
|
|
u.name === ownerValue || u.name === normalizedOwner;
|
|
this.logger.info('Comparing:', {
|
|
userId: u.id,
|
|
userEmail: u.email,
|
|
userName: u.name,
|
|
filterValue: ownerValue,
|
|
normalizedFilter: normalizedOwner,
|
|
idMatch: isIdMatch,
|
|
emailMatch: isEmailMatch,
|
|
nameMatch: isNameMatch,
|
|
});
|
|
return isIdMatch || isEmailMatch || isNameMatch;
|
|
});
|
|
this.logger.info('Looking for owner result:', {
|
|
ownerValue,
|
|
found: !!user,
|
|
userId: user?.id,
|
|
});
|
|
return user?.id || ownerValue;
|
|
})
|
|
.filter((id: string) => id);
|
|
|
|
if (ownerIds.length > 0) {
|
|
filters.owner = ownerIds;
|
|
}
|
|
}
|
|
|
|
const { data_assets, total } = await lastValueFrom(
|
|
this.catalogReadService.GetAllDataAssets(
|
|
{
|
|
search,
|
|
page,
|
|
size,
|
|
sort_by,
|
|
order,
|
|
filters: JSON.stringify(filters),
|
|
},
|
|
metadata,
|
|
),
|
|
);
|
|
|
|
console.log('MAESTRO RECEBEU RESPOSTA DO PI-FACTORY');
|
|
|
|
const result = JSON.parse(data_assets);
|
|
|
|
const response = await this.getAssetsUsersAndRoles(
|
|
result.data_assets,
|
|
customer_id,
|
|
);
|
|
|
|
return { data_assets: response, total };
|
|
}
|
|
|
|
async downloadAssets(
|
|
query: Record<string, any>,
|
|
metadata: Metadata,
|
|
customer_id: string,
|
|
) {
|
|
const data = await this.searchDataAssets(query, metadata, customer_id);
|
|
|
|
const formatData = data.data_assets.map((asset) => ({
|
|
id: asset.id,
|
|
display_name: asset.display_name,
|
|
data_asset_type: asset.data_asset_type,
|
|
created_at: asset.created_at,
|
|
tags: '[' + asset.tags.join(', ') + ']',
|
|
}));
|
|
|
|
const parser = ParserBuilder.build<AssetReporter>('csv');
|
|
|
|
const file = await parser.parse(formatData);
|
|
|
|
const timestamp = new Date().toISOString().replace(/[:.]/g, '-');
|
|
const filename = `dadosfera_assets_${timestamp}.csv`;
|
|
|
|
return {
|
|
file,
|
|
filename,
|
|
};
|
|
}
|
|
|
|
async getOneDataAsset(data: {
|
|
id: string;
|
|
customer_id: string;
|
|
metadata: Metadata;
|
|
}) {
|
|
const { customer_id, id, metadata } = data;
|
|
const { data_asset } = await lastValueFrom(
|
|
this.catalogReadService.GetOneDataAsset(
|
|
{ id, type: undefined },
|
|
metadata,
|
|
),
|
|
);
|
|
let asset = JSON.parse(data_asset);
|
|
asset = {
|
|
...asset,
|
|
p_roles: asset.roles,
|
|
p_users: asset.users,
|
|
};
|
|
asset = await this.getAssetsUsersAndRoles([asset], customer_id);
|
|
|
|
return { data_asset: asset[0] };
|
|
}
|
|
|
|
async getOneDataAssetByPipelineAndObject(data: {
|
|
customer_id: string;
|
|
pipeline: string;
|
|
object: string;
|
|
metadata: Metadata;
|
|
}) {
|
|
const { customer_id, metadata, pipeline, object } = data;
|
|
const { data_asset } = await lastValueFrom(
|
|
this.catalogReadService.GetOneDataAssetByPipelineAndObject(
|
|
{ pipeline, object },
|
|
metadata,
|
|
),
|
|
);
|
|
let asset = JSON.parse(data_asset);
|
|
asset = {
|
|
...asset,
|
|
p_roles: asset.roles,
|
|
p_users: asset.users,
|
|
};
|
|
asset = await this.getAssetsUsersAndRoles([asset], customer_id);
|
|
|
|
return { data_asset: asset[0] };
|
|
}
|
|
|
|
async updateCertificationStatus(data: {
|
|
data_asset_id: string;
|
|
body: IUpdateCertificationStatusRequest;
|
|
metadata: Metadata;
|
|
}) {
|
|
const { body, data_asset_id, metadata } = data;
|
|
|
|
await lastValueFrom(
|
|
this.catalogWriteService.UpdateDataAsset(
|
|
{
|
|
id: data_asset_id,
|
|
changes: JSON.stringify({
|
|
certification_status: body.certification_status,
|
|
}),
|
|
},
|
|
metadata,
|
|
),
|
|
);
|
|
|
|
return { certification_status: body.certification_status };
|
|
}
|
|
|
|
async updateOneDataAsset(data: {
|
|
data_asset_id: string;
|
|
customer_id: string;
|
|
body: IUpdateDataRequest;
|
|
metadata: Metadata;
|
|
}) {
|
|
const { body, customer_id, data_asset_id, metadata } = data;
|
|
|
|
const { data_asset } = await lastValueFrom(
|
|
this.catalogWriteService.UpdateDataAsset(
|
|
{ id: data_asset_id, changes: JSON.stringify(body) },
|
|
metadata,
|
|
),
|
|
);
|
|
let asset = JSON.parse(data_asset);
|
|
asset = {
|
|
...asset,
|
|
p_roles: asset.roles,
|
|
p_users: asset.users,
|
|
};
|
|
asset = await this.getAssetsUsersAndRoles([asset], customer_id);
|
|
|
|
return { data_asset: asset[0] };
|
|
}
|
|
|
|
async getDataDocs(id: string, assetType: string, metadata: Metadata) {
|
|
const { documentation } = await lastValueFrom(
|
|
this.catalogReadService.GetDatasetDoc({ id }, metadata),
|
|
);
|
|
|
|
const docs = JSON.parse(documentation);
|
|
return docs;
|
|
}
|
|
|
|
async getDatasetPreview(id: string, metadata: Metadata) {
|
|
const { preview } = await lastValueFrom(
|
|
this.catalogReadService.GetDatasetPreview(
|
|
{ id, type: undefined },
|
|
metadata,
|
|
),
|
|
);
|
|
const result = JSON.parse(preview);
|
|
return result;
|
|
}
|
|
|
|
async getDatasetColumnsMetadata(id: string, metadata: Metadata) {
|
|
const { columns_metadata } = await lastValueFrom(
|
|
this.catalogReadService.GetDatasetColumnsMetadata(
|
|
{ id, type: undefined },
|
|
metadata,
|
|
),
|
|
);
|
|
const result = JSON.parse(columns_metadata);
|
|
return result;
|
|
}
|
|
|
|
async createDataDocs(body: CreateDataDocsDTO, metadata: Metadata) {
|
|
if (body.asset_type === 'table' || body.asset_type === 'view') {
|
|
return this.createDataDocsViaNimbus(body);
|
|
}
|
|
|
|
return this.createDataDocsViaGrpc(body, metadata);
|
|
}
|
|
|
|
private async createDataDocsViaNimbus(body: CreateDataDocsDTO) {
|
|
this.logger.info('Creating data docs via Nimbus for table/view');
|
|
const nimbusUrl = this._getNimbusUrl(body);
|
|
const { data } = await axios.post(
|
|
`${nimbusUrl}/api/catalog/data-docs/`,
|
|
body,
|
|
);
|
|
return data;
|
|
}
|
|
|
|
private async createDataDocsViaGrpc(body: CreateDataDocsDTO, metadata: Metadata) {
|
|
this.logger.info('Creating data docs via gRPC for other asset types');
|
|
try {
|
|
const response: any = await lastValueFrom(
|
|
this.catalogWriteService.UpdateDataAssetDoc(
|
|
{
|
|
id: body.table_id,
|
|
docs: body.docs,
|
|
},
|
|
metadata,
|
|
),
|
|
);
|
|
|
|
return response;
|
|
} catch (error) {
|
|
this.logger.error('Error creating data asset docs:', error);
|
|
throw new HttpException(
|
|
'Failed to create data asset documentation',
|
|
HttpStatus.INTERNAL_SERVER_ERROR,
|
|
);
|
|
}
|
|
}
|
|
|
|
async findAllTags(data, metadata) {
|
|
this.logger.info('CatalogService - findAllCustomerTags');
|
|
|
|
const response = await lastValueFrom(
|
|
this.catalogReadService.GetCustomerTags(data, metadata),
|
|
)
|
|
.then((res) => {
|
|
this.logger.info('Done');
|
|
return res;
|
|
})
|
|
.catch((err) => {
|
|
this.logger.error(err.message);
|
|
throw new Error(err);
|
|
});
|
|
|
|
return response;
|
|
}
|
|
|
|
async findSchemas(metadata: Metadata) {
|
|
this.logger.info('CatalogService - findSchemas');
|
|
|
|
try {
|
|
const response = await lastValueFrom(
|
|
this.catalogReadService.GetSchemas({}, metadata),
|
|
);
|
|
|
|
return response;
|
|
|
|
} catch (error) {
|
|
this.logger.error('Error fetching schemas:', error);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async getAssetsUsersAndRoles(data_assets: Array<any>, customer_id: string) {
|
|
const { users: customer_users } =
|
|
await this.userService.findAllUsersByCustomerId(customer_id);
|
|
const { roles: customer_roles } = await this.roleService.roleSearch(
|
|
{},
|
|
{ customer_id },
|
|
);
|
|
return data_assets.map((data_asset) => {
|
|
const owner = customer_users.find(
|
|
(u) => u.id === data_asset.owner,
|
|
)?.email;
|
|
|
|
const roles = [];
|
|
const users = [];
|
|
const data_asset_roles = data_asset?.roles || [];
|
|
for (const role_id of data_asset_roles) {
|
|
const role = customer_roles.find((r) => r.id === role_id);
|
|
if (role) roles.push({ id: role.id, name: role.name });
|
|
}
|
|
|
|
const data_asset_users = data_asset?.users || [];
|
|
for (const user_id of data_asset_users) {
|
|
const user = customer_users.find((r) => r.id === user_id);
|
|
if (user) users.push({ id: user.id, email: user.email });
|
|
}
|
|
return {
|
|
...data_asset,
|
|
roles,
|
|
users,
|
|
owner,
|
|
} as typeof data_asset;
|
|
});
|
|
}
|
|
|
|
async triggerCatalog(data: TriggerCatalogReq, metadata: Metadata) {
|
|
const { session } = await lastValueFrom(
|
|
this.catalogWriteService.TriggerDatasetCataloging(data, metadata),
|
|
);
|
|
return session;
|
|
}
|
|
async getDatasetCatalogTask(session: string, metadata: Metadata) {
|
|
const res = await lastValueFrom(
|
|
this.catalogReadService.GetDatasetCatalogTask({ session }, metadata),
|
|
);
|
|
return res;
|
|
}
|
|
|
|
async addRlsRule(data: AddRlsRuleRequest, metadata: Metadata) {
|
|
const res = await lastValueFrom(
|
|
this.catalogWriteService.AddRlsRule(data, metadata),
|
|
);
|
|
return res;
|
|
}
|
|
|
|
async removeRlsRule(id: number, metadata: Metadata) {
|
|
const res = await lastValueFrom(
|
|
this.catalogWriteService.RemoveRlsRule({ id }, metadata),
|
|
);
|
|
return res;
|
|
}
|
|
|
|
async batchRemoveRlsRule(
|
|
query: BatchRemoveRlsRulesRequest,
|
|
metadata: Metadata,
|
|
) {
|
|
const { id_rls, nimbus_dashboard_id } = query;
|
|
|
|
if (id_rls && nimbus_dashboard_id) {
|
|
throw new BadRequestException(
|
|
"You can't delete using both parameters. Choose either 'id_rls' or 'nimbus_dashboard_id'",
|
|
);
|
|
}
|
|
if (id_rls) {
|
|
await lastValueFrom(
|
|
this.catalogWriteService.RemoveRlsRulesByRlsId({ id_rls }, metadata),
|
|
);
|
|
} else if (nimbus_dashboard_id) {
|
|
await lastValueFrom(
|
|
this.catalogWriteService.RemoveRlsRulesByDashboardId(
|
|
{ nimbus_dashboard_id: parseInt(nimbus_dashboard_id) },
|
|
metadata,
|
|
),
|
|
);
|
|
}
|
|
return 'OK';
|
|
}
|
|
|
|
async getRlsRules(data: GetRlsRulesRequest, metadata: Metadata) {
|
|
const res = await lastValueFrom(
|
|
this.catalogReadService.GetRlsRules(data, metadata),
|
|
);
|
|
return res.rls_rules;
|
|
}
|
|
|
|
async getOneRlsRule(id: number, metadata: Metadata) {
|
|
const res = await lastValueFrom(
|
|
this.catalogReadService.GetOneRlsRule({ id }, metadata),
|
|
);
|
|
return res.rls_rule;
|
|
}
|
|
|
|
async getNimbusDashboards(
|
|
data: GetNimbusDashboardsRequest,
|
|
metadata: Metadata,
|
|
) {
|
|
const res = await lastValueFrom(
|
|
this.catalogReadService.GetNimbusDashboards(data, metadata),
|
|
);
|
|
return res.dashboards;
|
|
}
|
|
|
|
async createTableMetadata(body: any): Promise<number> {
|
|
const nimbusUrl = this._getNimbusUrl(body);
|
|
this.logger.info(`Nimbus URL: ${nimbusUrl}`, { ...body.logMetadata });
|
|
|
|
const endpoint = `${nimbusUrl}/api/catalog/table-metadata/`;
|
|
|
|
this.logger.info(
|
|
`Creating table metadata for table ${body.table_metadata.table_name}`,
|
|
{ ...body.logMetadata },
|
|
);
|
|
this.logger.info(`Using endpoint: ${endpoint}`, { ...body.logMetadata });
|
|
this.logger.debug(`Payload: ${JSON.stringify(body.table_metadata)}`, {
|
|
...body.logMetadata,
|
|
});
|
|
|
|
try {
|
|
const { data, status } = await axios.post(endpoint, {
|
|
...body.table_metadata,
|
|
});
|
|
|
|
this.logger.info(
|
|
`Table metadata created successfully with status ${status} for table ${body.table_metadata.table_name}`,
|
|
{ ...body.logMetadata },
|
|
);
|
|
return data.id;
|
|
} catch (error) {
|
|
this.logger.error(
|
|
`Failed to create table metadata for table ${body.table_metadata.table_name} failed with status ${
|
|
error.response?.status
|
|
} because of ${JSON.stringify(error.response?.data) || error.message}`,
|
|
{ ...body.logMetadata },
|
|
);
|
|
throw new Error(error.response?.data?.message || error.message);
|
|
}
|
|
}
|
|
|
|
async createColumnMetadata(body: any): Promise<number[]> {
|
|
const nimbusUrl = this._getNimbusUrl(body);
|
|
this.logger.info(`Nimbus URL: ${nimbusUrl}`, body.logMetadata);
|
|
const endpoint = `${nimbusUrl}/api/catalog/column-metadata/`;
|
|
|
|
try {
|
|
this.logger.info(
|
|
`Creating column metadata for table ${body.column_metadata.table_name}`,
|
|
{ ...body.logMetadata },
|
|
);
|
|
this.logger.info(`Using endpoint: ${endpoint}`, { ...body.logMetadata });
|
|
this.logger.debug(
|
|
`Payload: ${JSON.stringify(body.column_metadata)}`,
|
|
{ ...body.logMetadata },
|
|
);
|
|
const { data, status } = await axios.post(
|
|
endpoint,
|
|
body.column_metadata,
|
|
);
|
|
|
|
this.logger.info(
|
|
`Column metadata created successfully with status ${status} for table ${body.column_metadata.table_name}`,
|
|
{ ...body.logMetadata },
|
|
);
|
|
return data.map((column) => column.id);
|
|
} catch (error) {
|
|
this.logger.error(
|
|
`Failed to create column metadata failed with status for table ${body.column_metadata.table_name} ${
|
|
error.response?.status
|
|
} because of ${error.response?.data || error.message}`,
|
|
{ ...body.logMetadata },
|
|
);
|
|
throw new Error(error.response?.data?.message || error.message);
|
|
}
|
|
}
|
|
|
|
async createDataPreview(body: any): Promise<number> {
|
|
const nimbusUrl = this._getNimbusUrl(body);
|
|
this.logger.info(`Nimbus URL: ${nimbusUrl}`, { ...body.logMetadata });
|
|
const endpoint = `${nimbusUrl}/api/catalog/data-preview/`;
|
|
|
|
this.logger.info(
|
|
`Creating data preview for table ${body.data_preview.table_name}`,
|
|
{ ...body.logMetadata },
|
|
);
|
|
this.logger.info(`Using endpoint: ${endpoint}`, { ...body.logMetadata });
|
|
this.logger.debug(
|
|
`Payload: ${JSON.stringify(body.data_preview)}`,
|
|
{ ...body.logMetadata },
|
|
);
|
|
|
|
try {
|
|
const { data, status } = await axios.post(endpoint, body.data_preview);
|
|
|
|
this.logger.info(
|
|
`Data preview created successfully with status ${status} for table ${body.data_preview.table_name}`,
|
|
{ ...body.logMetadata },
|
|
);
|
|
return data.id;
|
|
} catch (error) {
|
|
this.logger.error(
|
|
`Failed to create data preview for table ${body.data_preview.table_name} failed with status ${
|
|
error.response?.status
|
|
} because of ${error.response?.data || error.message}`,
|
|
{ ...body.logMetadata },
|
|
);
|
|
throw new Error(error.response?.data?.message || error.message);
|
|
}
|
|
}
|
|
|
|
async renameTableOnNimbus(
|
|
nimbusUrl: string,
|
|
nimbusId: number,
|
|
changes: { table_name?: string; table_schema?: string; display_name?: string },
|
|
): Promise<void> {
|
|
const endpoint = `${nimbusUrl}/api/catalog/table-metadata/${nimbusId}`;
|
|
this.logger.info(`Renaming table-metadata ${nimbusId} on Nimbus`, { endpoint, changes });
|
|
await axios.patch(endpoint, changes);
|
|
}
|
|
|
|
async renameColumnMetadataOnNimbus(
|
|
nimbusUrl: string,
|
|
databaseName: string,
|
|
oldTableName: string,
|
|
oldTableSchema: string,
|
|
newTableName: string,
|
|
newTableSchema: string,
|
|
): Promise<void> {
|
|
const listEndpoint = `${nimbusUrl}/api/catalog/column-metadata/?database_name=${encodeURIComponent(databaseName)}&table_name=${encodeURIComponent(oldTableName)}&table_schema=${encodeURIComponent(oldTableSchema)}`;
|
|
this.logger.info(`Fetching column-metadata records to rename`, { listEndpoint });
|
|
const { data: columns } = await axios.get(listEndpoint);
|
|
|
|
const filtered = Array.isArray(columns) ? columns : [];
|
|
|
|
for (const column of filtered) {
|
|
const patchEndpoint = `${nimbusUrl}/api/catalog/column-metadata/${column.id}`;
|
|
await axios.patch(patchEndpoint, {
|
|
table_name: newTableName,
|
|
table_schema: newTableSchema,
|
|
});
|
|
}
|
|
this.logger.info(`Renamed ${filtered.length} column-metadata records on Nimbus`);
|
|
}
|
|
|
|
async renameDataPreviewOnNimbus(
|
|
nimbusUrl: string,
|
|
databaseName: string,
|
|
oldTableName: string,
|
|
oldTableSchema: string,
|
|
newTableName: string,
|
|
newTableSchema: string,
|
|
): Promise<void> {
|
|
const listEndpoint = `${nimbusUrl}/api/catalog/data-preview/?database_name=${encodeURIComponent(databaseName)}&table_name=${encodeURIComponent(oldTableName)}&table_schema=${encodeURIComponent(oldTableSchema)}`;
|
|
this.logger.info(`Fetching data-preview records to rename`, { listEndpoint });
|
|
const { data: previews } = await axios.get(listEndpoint);
|
|
|
|
const filtered = Array.isArray(previews) ? previews : [];
|
|
|
|
for (const preview of filtered) {
|
|
const patchEndpoint = `${nimbusUrl}/api/catalog/data-preview/${preview.id}`;
|
|
await axios.patch(patchEndpoint, {
|
|
table_name: newTableName,
|
|
table_schema: newTableSchema,
|
|
});
|
|
}
|
|
this.logger.info(`Renamed ${filtered.length} data-preview records on Nimbus`);
|
|
}
|
|
|
|
async catalogDatasetItem(table_metadata_id: number, metadata: Metadata) {
|
|
const customer_name_raw = metadata.get('customer_name');
|
|
|
|
const customer_name = customer_name_raw?.[0]?.toString();
|
|
if (!customer_name) {
|
|
throw new BadRequestException('Customer name not found in metadata');
|
|
}
|
|
const res = await lastValueFrom(
|
|
this.platformWriteService.CatalogDataAssets(
|
|
{
|
|
data_assets: [
|
|
{
|
|
data_asset_id: table_metadata_id.toString(),
|
|
customer_name: customer_name,
|
|
data_asset_type: 'dataset',
|
|
},
|
|
],
|
|
},
|
|
metadata,
|
|
),
|
|
);
|
|
return res;
|
|
}
|
|
}
|
|
|
|
export { CatalogService };
|