* feat(plugin-workflow): add aggregate instruction * test(plugin-workflow): add test cases * fix(plugin-workflow): fix types * fix(plugin-workflow): fix double result type * test(plugin-workflow): fix test cases in mysql * refactor(plugin-workflow): consolidate variables api * fix(plugin-workflow): fix create node variable * fix(plugin-workflow): fix aggregate association name * fix(plugin-workflow): fix test cases * fix(plugin-workflow): fix aggregate node config for duplication * fix(plugin-workflow): fix variable api * fix(plugin-workflow): fix variable api caller * fix(plugin-workflow): fix job button style
44 lines
1.3 KiB
TypeScript
44 lines
1.3 KiB
TypeScript
import { DataTypes } from 'sequelize';
|
|
|
|
import FlowNodeModel from '../models/FlowNode';
|
|
import Processor from '../Processor';
|
|
import { JOB_STATUS } from '../constants';
|
|
import { BelongsToManyRepository, HasManyRepository } from '@nocobase/database';
|
|
|
|
const aggregators = {
|
|
count: 'count',
|
|
sum: 'sum',
|
|
avg: 'avg',
|
|
min: 'min',
|
|
max: 'max',
|
|
};
|
|
|
|
export default {
|
|
async run(node: FlowNodeModel, input, processor: Processor) {
|
|
const { aggregator, associated, collection, association = {}, params = {} } = node.config;
|
|
const options = processor.getParsedValue(params);
|
|
const { database } = <typeof FlowNodeModel>node.constructor;
|
|
const repo = associated
|
|
? database.getRepository<HasManyRepository | BelongsToManyRepository>(
|
|
`${association?.associatedCollection}.${association.name}`,
|
|
processor.getParsedValue(association?.associatedKey),
|
|
)
|
|
: database.getRepository(collection);
|
|
|
|
if (!options.dataType && aggregator === 'avg') {
|
|
options.dataType = DataTypes.DOUBLE;
|
|
}
|
|
|
|
const result = await repo.aggregate({
|
|
...options,
|
|
method: aggregators[aggregator],
|
|
transaction: processor.transaction,
|
|
});
|
|
|
|
return {
|
|
result: options.dataType === DataTypes.DOUBLE ? Number(result) : result,
|
|
status: JOB_STATUS.RESOLVED,
|
|
};
|
|
},
|
|
};
|