admin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Upsert endpoint and CSV import upsert (#5970)

Browse Source 
This PR introduces an `upsert` parameter (along the existing `data`
param) for `createOne` and `createMany` mutations.

When upsert is set to `true`, the function will look for records with
the same id if an id was passed. If not id was passed, it will leverage
the existing duplicate check mechanism to find a duplicate. If a record
is found, then the function will perform an update instead of a create.

Unfortunately I had to remove some nice tests that existing on the args
factory. Those tests where mostly testing the duplication rule
generation logic but through a GraphQL angle. Since I moved the
duplication rule logic to a dedicated service, if I kept the tests but
mocked the service we wouldn't really be testing anything useful. The
right path would be to create new tests for this service that compare
the JSON output and not the GraphQL output but I chose not to work on
this as it's equivalent to rewriting the tests from scratch and I have
other competing priorities.
This commit is contained in:
Félix Malfait
2024-06-26 11:39:16 +02:00
committed by GitHub
parent 1736aee7ff
commit cf67ed09d0
43 changed files with 609 additions and 531 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
packages/twenty-server/src/engine/api/graphql/workspace-query-runner/factories/__tests__/query-runner-args.factory.spec.ts
View File

@ -152,8 +152,8 @@ describe('QueryRunnerArgsFactory', () => {
} as WorkspaceQueryRunnerOptions;
const args = {
id: '123',
data: { testNumber: '1', otherField: 'test' },
ids: ['123'],
data: [{ testNumber: '1', otherField: 'test' }],
};
const result = await factory.create(
@ -163,8 +163,8 @@ describe('QueryRunnerArgsFactory', () => {
);
expect(result).toEqual({
id: 123,
data: { testNumber: 1, otherField: 'test' },
ids: [123],
data: [{ testNumber: 1, position: 2, otherField: 'test' }],
});
});
});

35
packages/twenty-server/src/engine/api/graphql/workspace-query-runner/factories/query-runner-args.factory.ts
View File

@ -10,7 +10,10 @@ import {
ResolverArgs,
ResolverArgsType,
} from 'src/engine/api/graphql/workspace-resolver-builder/interfaces/workspace-resolvers-builder.interface';
import { RecordFilter } from 'src/engine/api/graphql/workspace-query-builder/interfaces/record.interface';
import {
Record,
RecordFilter,
} from 'src/engine/api/graphql/workspace-query-builder/interfaces/record.interface';
import { FieldMetadataType } from 'src/engine/metadata-modules/field-metadata/field-metadata.entity';
import { hasPositionField } from 'src/engine/metadata-modules/object-metadata/utils/has-position-field.util';
@ -47,12 +50,12 @@ export class QueryRunnerArgsFactory {
return {
...args,
data: await Promise.all(
(args as CreateManyResolverArgs).data.map((arg, index) =>
(args as CreateManyResolverArgs).data?.map((arg, index) =>
this.overrideDataByFieldMetadata(arg, options, fieldMetadataMap, {
argIndex: index,
shouldBackfillPosition,
}),
),
) ?? [],
),
} satisfies CreateManyResolverArgs;
case ResolverArgsType.FindOne:
@ -75,25 +78,27 @@ export class QueryRunnerArgsFactory {
case ResolverArgsType.FindDuplicates:
return {
...args,
id: await this.overrideValueByFieldMetadata(
'id',
(args as FindDuplicatesResolverArgs).id,
fieldMetadataMap,
ids: (await Promise.all(
(args as FindDuplicatesResolverArgs).ids?.map((id) =>
this.overrideValueByFieldMetadata('id', id, fieldMetadataMap),
) ?? [],
)) as string[],
data: await Promise.all(
(args as FindDuplicatesResolverArgs).data?.map((arg, index) =>
this.overrideDataByFieldMetadata(arg, options, fieldMetadataMap, {
argIndex: index,
shouldBackfillPosition,
}),
) ?? [],
),
data: await this.overrideDataByFieldMetadata(
(args as FindDuplicatesResolverArgs).data,
options,
fieldMetadataMap,
{ shouldBackfillPosition: false },
),
};
} satisfies FindDuplicatesResolverArgs;
default:
return args;
}
}
private async overrideDataByFieldMetadata(
data: Record<string, any> | undefined,
data: Partial<Record> | undefined,
options: WorkspaceQueryRunnerOptions,
fieldMetadataMap: Map<string, FieldMetadataInterface>,
argPositionBackfillInput: ArgPositionBackfillInput,

2
packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-runner.module.ts
View File

@ -10,6 +10,7 @@ import { ObjectMetadataRepositoryModule } from 'src/engine/object-metadata-repos
import { TelemetryListener } from 'src/engine/api/graphql/workspace-query-runner/listeners/telemetry.listener';
import { AnalyticsModule } from 'src/engine/core-modules/analytics/analytics.module';
import { RecordPositionBackfillCommand } from 'src/engine/api/graphql/workspace-query-runner/commands/0-20-record-position-backfill.command';
import { DuplicateModule } from 'src/engine/core-modules/duplicate/duplicate.module';
import { WorkspaceQueryRunnerService } from './workspace-query-runner.service';
@ -23,6 +24,7 @@ import { EntityEventsToDbListener } from './listeners/entity-events-to-db.listen
WorkspaceQueryHookModule,
ObjectMetadataRepositoryModule.forFeature([WorkspaceMemberWorkspaceEntity]),
AnalyticsModule,
DuplicateModule,
],
providers: [
WorkspaceQueryRunnerService,

120
packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-runner.service.ts
View File

@ -52,6 +52,7 @@ import { STANDARD_OBJECT_IDS } from 'src/engine/workspace-manager/workspace-sync
import { assertIsValidUuid } from 'src/engine/api/graphql/workspace-query-runner/utils/assert-is-valid-uuid.util';
import { isQueryTimeoutError } from 'src/engine/utils/query-timeout.util';
import { InjectMessageQueue } from 'src/engine/integrations/message-queue/decorators/message-queue.decorator';
import { DuplicateService } from 'src/engine/core-modules/duplicate/duplicate.service';
import { WorkspaceQueryRunnerOptions } from './interfaces/query-runner-option.interface';
import {
@ -77,6 +78,7 @@ export class WorkspaceQueryRunnerService {
private readonly eventEmitter: EventEmitter2,
private readonly workspaceQueryHookService: WorkspaceQueryHookService,
private readonly environmentService: EnvironmentService,
private readonly duplicateService: DuplicateService,
) {}
async findMany<
@ -167,16 +169,16 @@ export class WorkspaceQueryRunnerService {
}
async findDuplicates<TRecord extends IRecord = IRecord>(
args: FindDuplicatesResolverArgs<TRecord>,
args: FindDuplicatesResolverArgs<Partial<TRecord>>,
options: WorkspaceQueryRunnerOptions,
): Promise<IConnection<TRecord> | undefined> {
if (!args.data && !args.id) {
if (!args.data && !args.ids) {
throw new BadRequestException(
'You have to provide either "data" or "id" argument',
);
}
if (!args.id && isEmpty(args.data)) {
if (!args.ids && isEmpty(args.data)) {
throw new BadRequestException(
'The "data" condition can not be empty when ID input not provided',
);
@ -190,37 +192,24 @@ export class WorkspaceQueryRunnerService {
ResolverArgsType.FindDuplicates,
)) as FindDuplicatesResolverArgs<TRecord>;
let existingRecord: Record<string, unknown> | undefined;
let existingRecords: IRecord[] | undefined = undefined;
if (computedArgs.id) {
const existingRecordQuery =
this.workspaceQueryBuilderFactory.findDuplicatesExistingRecord(
computedArgs.id,
options,
);
const existingRecordResult = await this.execute(
existingRecordQuery,
if (computedArgs.ids && computedArgs.ids.length > 0) {
existingRecords = await this.duplicateService.findExistingRecords(
computedArgs.ids,
objectMetadataItem,
workspaceId,
);
const parsedResult = await this.parseResult<Record<string, unknown>>(
existingRecordResult,
objectMetadataItem,
'',
);
existingRecord = parsedResult?.edges?.[0]?.node;
if (!existingRecord) {
throw new NotFoundError(`Object with id ${args.id} not found`);
if (!existingRecords || existingRecords.length === 0) {
throw new NotFoundError(`Object with id ${args.ids} not found`);
}
}
const query = await this.workspaceQueryBuilderFactory.findDuplicates(
computedArgs,
options,
existingRecord,
existingRecords,
);
await this.workspaceQueryHookService.executePreQueryHooks(
@ -237,17 +226,22 @@ export class WorkspaceQueryRunnerService {
result,
objectMetadataItem,
'',
true,
);
}
async createMany<Record extends IRecord = IRecord>(
args: CreateManyResolverArgs<Record>,
args: CreateManyResolverArgs<Partial<Record>>,
options: WorkspaceQueryRunnerOptions,
): Promise<Record[] | undefined> {
const { workspaceId, userId, objectMetadataItem } = options;
assertMutationNotOnRemoteObject(objectMetadataItem);
if (args.upsert) {
return await this.upsertMany(args, options);
}
args.data.forEach((record) => {
if (record?.id) {
assertIsValidUuid(record.id);
@ -305,17 +299,73 @@ export class WorkspaceQueryRunnerService {
return parsedResults;
}
async upsertMany<Record extends IRecord = IRecord>(
args: CreateManyResolverArgs<Partial<Record>>,
options: WorkspaceQueryRunnerOptions,
): Promise<Record[] | undefined> {
const ids = args.data
.map((item) => item.id)
.filter((id) => id !== undefined);
const existingRecords =
ids.length > 0
? await this.duplicateService.findExistingRecords(
ids as string[],
options.objectMetadataItem,
options.workspaceId,
)
: [];
const existingRecordsMap = new Map(
existingRecords.map((record) => [record.id, record]),
);
const results: Record[] = [];
const recordsToCreate: Partial<Record>[] = [];
for (const payload of args.data) {
if (payload.id && existingRecordsMap.has(payload.id)) {
const result = await this.updateOne(
{ id: payload.id, data: payload },
options,
);
if (result) {
results.push(result);
}
} else {
recordsToCreate.push(payload);
}
}
if (recordsToCreate.length > 0) {
const createResults = await this.createMany(
{ data: recordsToCreate } as CreateManyResolverArgs<Partial<Record>>,
options,
);
if (createResults) {
results.push(...createResults);
}
}
return results;
}
async createOne<Record extends IRecord = IRecord>(
args: CreateOneResolverArgs<Record>,
args: CreateOneResolverArgs<Partial<Record>>,
options: WorkspaceQueryRunnerOptions,
): Promise<Record | undefined> {
const results = await this.createMany({ data: [args.data] }, options);
const results = await this.createMany(
{ data: [args.data], upsert: args.upsert },
options,
);
return results?.[0];
}
async updateOne<Record extends IRecord = IRecord>(
args: UpdateOneResolverArgs<Record>,
args: UpdateOneResolverArgs<Partial<Record>>,
options: WorkspaceQueryRunnerOptions,
): Promise<Record | undefined> {
const { workspaceId, userId, objectMetadataItem } = options;
@ -373,7 +423,7 @@ export class WorkspaceQueryRunnerService {
}
async updateMany<Record extends IRecord = IRecord>(
args: UpdateManyResolverArgs<Record>,
args: UpdateManyResolverArgs<Partial<Record>>,
options: WorkspaceQueryRunnerOptions,
): Promise<Record[] | undefined> {
const { userId, workspaceId, objectMetadataItem } = options;
@ -609,11 +659,21 @@ export class WorkspaceQueryRunnerService {
graphqlResult: PGGraphQLResult | undefined,
objectMetadataItem: ObjectMetadataInterface,
command: string,
isMultiQuery = false,
): Promise<Result> {
const entityKey = `${command}${computeObjectTargetTable(
objectMetadataItem,
)}Collection`;
const result = graphqlResult?.[0]?.resolve?.data?.[entityKey];
const result = !isMultiQuery
? graphqlResult?.[0]?.resolve?.data?.[entityKey]
: Object.keys(graphqlResult?.[0]?.resolve?.data).reduce(
(acc: IRecord[], dataItem, index) => {
acc.push(graphqlResult?.[0]?.resolve?.data[`${entityKey}${index}`]);
return acc;
},
[],
);
const errors = graphqlResult?.[0]?.resolve?.errors;
if (