Add functions to create single incremental/delete reader (#406)

Author: eskrm

src/graphql/graphql.ts

@@ -943,8 +943,8 @@ interface IncrementalReadersConfig {
   readonly graph: string;
   readonly pageSize: number;
   readonly graphSchema: gql.GraphQLSchema;
-  readonly avoidCollisions: boolean;
-  readonly scalarsOnly: boolean;
+  readonly avoidCollisions?: boolean;
+  readonly scalarsOnly?: boolean;
 }
 
 export function createIncrementalReadersV2(
@@ -970,6 +970,74 @@ export function createIncrementalReadersV2(
   return result;
 }
 
+export interface IncrementalReaderConfig {
+  readonly model: string;
+  readonly client: FarosClient;
+  readonly graph: string;
+  readonly graphSchema: gql.GraphQLSchema;
+  readonly pageSize: number;
+  readonly avoidCollisions: boolean;
+  readonly scalarsOnly: boolean;
+}
+
+export function createIncrementalReader(
+  cfg: IncrementalReaderConfig
+): Reader | undefined {
+  const type = cfg.graphSchema.getType(cfg.model);
+  if (isV2ModelType(type)) {
+    const avoidCollisions = cfg.avoidCollisions ?? true;
+    const scalarsOnly = cfg.scalarsOnly ?? false;
+    return readerFromQuery({
+      client: cfg.client,
+      graph: cfg.graph,
+      graphSchema: cfg.graphSchema,
+      pageSize: cfg.pageSize,
+      incremental: true,
+      query: buildIncrementalQueryV2({
+        type,
+        avoidCollisions,
+        scalarsOnly,
+      }),
+    });
+  }
+  return undefined;
+}
+
+export interface DeleteReaderConfig {
+  readonly model: string;
+  readonly client: FarosClient;
+  readonly graph: string;
+  readonly graphSchema: gql.GraphQLSchema;
+  readonly pageSize: number;
+}
+
+export function createDeleteReader(
+  cfg: DeleteReaderConfig
+): Reader | undefined {
+  const type = cfg.graphSchema.getType(cfg.model);
+  if (!isV2ModelType(type)) {
+    return undefined;
+  }
+
+  const deleteQuery = `
+    query delete($from: timestamptz!, $to: timestamptz!) {
+      ${cfg.model}_history(where: {_and: [
+        {action: {_eq: "delete"}},
+        {actionAt: {_gte: $from, _lt: $to}}
+      ]}) {
+        id
+      }
+    }`;
+  const [deleteReader] = createNonIncrementalReaders({
+    client: cfg.client,
+    graph: cfg.graph,
+    pageSize: cfg.pageSize,
+    graphSchema: cfg.graphSchema,
+    queries: [{gql: deleteQuery, name: cfg.model}],
+  });
+  return deleteReader;
+}
+
 interface IncrementalQueriesConfig {
   readonly graphSchema: gql.GraphQLSchema;
   readonly primaryKeys?: Dictionary<ReadonlyArray<string>>;

src/index.ts

@@ -46,12 +46,16 @@ export {
 } from './graphql/query-builder';
 export {
   AnyRecord,
+  DeleteReaderConfig,
   FlattenContext,
+  IncrementalReaderConfig,
   PaginatedQuery,
   Reader,
   RecordIterable,
   buildIncrementalQueryV2,
+  createDeleteReader,
   createIncrementalQueriesV2,
+  createIncrementalReader,
   createIncrementalReadersV2,
   createNonIncrementalReaders,
   crossMerge,

test/graphql.test.ts

@@ -2,6 +2,7 @@ import 'jest-extended';
 
 import * as gql from 'graphql';
 
+import {FarosClient} from '../src/client';
 import * as sut from '../src/graphql/graphql';
 import {
   graphSchemaV2,
@@ -442,4 +443,37 @@ describe('graphql', () => {
       path: ['cicd_Build'],
     });
   });
+
+  test('create incremental reader', () => {
+    const reader = sut.createIncrementalReader({
+      model: 'cicd_Build',
+      client: {} as unknown as FarosClient,
+      graph: 'graph',
+      graphSchema: graphSchemaV2,
+      pageSize: 1,
+      avoidCollisions: false,
+      scalarsOnly: true,
+    });
+
+    expect(reader?.metadata).toMatchObject({
+      name: 'cicd_Build',
+      modelKeys: ['id'],
+      incremental: true,
+    });
+  });
+
+  test('create delete reader', () => {
+    const reader = sut.createDeleteReader({
+      model: 'cicd_Build',
+      client: {} as unknown as FarosClient,
+      graph: 'graph',
+      graphSchema: graphSchemaV2,
+      pageSize: 1,
+    });
+
+    expect(reader?.metadata).toMatchObject({
+      name: 'cicd_Build',
+      incremental: false,
+    });
+  });
 });

test/resources/schemas/schema-v2.gql

@@ -54,6 +54,14 @@ type query_root {
     """limit the number of rows returned"""
     limit: Int
   ): [cicd_Artifact!]!
+
+  """
+  fetch data from the table: "cicd_Build_history"
+  """
+  cicd_Build_history(
+    """limit the number of rows returned"""
+    limit: Int
+  ): [cicd_Build_history!]!
 }
 
 type ims_Incident {
@@ -286,3 +294,34 @@ type cicd_Artifact {
   uid: String!
 }
 
+"""
+columns and relationships of "cicd_Build_history"
+"""
+type cicd_Build_history {
+  action: String
+  actionAt: timestamptz!
+  actionBy: String
+  agentId: String
+  createdAt: timestamptz
+  endedAt: timestamptz
+  environmentVariables(
+    """JSON select path"""
+    path: String
+  ): jsonb
+  id: String!
+  isPhantom: Boolean
+  name: String
+  number: Int
+  origin: String
+  pipelineId: String
+  refreshedAt: timestamptz!
+  startedAt: timestamptz
+  status(
+    """JSON select path"""
+    path: String
+  ): jsonb
+  statusCategory: String
+  statusDetail: String
+  uid: String!
+  url: String
+}