From 0e9ec444b7f0e453151c920b1e144429fbd78976 Mon Sep 17 00:00:00 2001 From: Hellgren Heikki Date: Thu, 5 Jun 2025 15:41:59 +0300 Subject: [PATCH 1/6] feat: allow streaming catalog entities Signed-off-by: Hellgren Heikki --- .changeset/olive-moons-burn.md | 21 ++++++ .../catalog-client/report-testUtils.api.md | 3 + packages/catalog-client/report.api.md | 14 ++++ .../catalog-client/src/CatalogClient.test.ts | 65 +++++++++++++++++++ packages/catalog-client/src/CatalogClient.ts | 27 ++++++++ .../testUtils/InMemoryCatalogClient.test.ts | 10 +++ .../src/testUtils/InMemoryCatalogClient.ts | 17 +++++ packages/catalog-client/src/types/api.ts | 24 +++++++ packages/catalog-client/src/types/index.ts | 1 + plugins/catalog-node/report-testUtils.api.md | 6 ++ plugins/catalog-node/report.api.md | 6 ++ plugins/catalog-node/src/catalogService.ts | 22 ++++++- .../src/testUtils/catalogServiceMock.ts | 3 +- plugins/catalog-node/src/testUtils/types.ts | 6 ++ .../src/testUtils/catalogApiMock.ts | 1 + 15 files changed, 222 insertions(+), 4 deletions(-) create mode 100644 .changeset/olive-moons-burn.md diff --git a/.changeset/olive-moons-burn.md b/.changeset/olive-moons-burn.md new file mode 100644 index 0000000000..feeff93d3d --- /dev/null +++ b/.changeset/olive-moons-burn.md @@ -0,0 +1,21 @@ +--- +'@backstage/catalog-client': minor +'@backstage/plugin-catalog-react': minor +'@backstage/plugin-catalog-node': minor +--- + +Introduced new `streamEntities` async generator method for the catalog. + +Catalog API and Catalog Service now includes a `streamEntities` method that allows for streaming entities from the catalog. +This method is designed to handle large datasets efficiently by processing entities in a stream rather than loading them +all into memory at once. This is useful when you need to fetch a large number of entities but do not want to use pagination +or fetch all entities at once. + +Example usage: + +```ts +const stream = catalogClient.streamEntities({}, { token }); +for await (const entity of stream) { + // Handle entity +} +``` diff --git a/packages/catalog-client/report-testUtils.api.md b/packages/catalog-client/report-testUtils.api.md index f5bcf55c53..26e79c9f96 100644 --- a/packages/catalog-client/report-testUtils.api.md +++ b/packages/catalog-client/report-testUtils.api.md @@ -22,6 +22,7 @@ import { GetLocationsResponse } from '@backstage/catalog-client'; import { Location as Location_2 } from '@backstage/catalog-client'; import { QueryEntitiesRequest } from '@backstage/catalog-client'; import { QueryEntitiesResponse } from '@backstage/catalog-client'; +import { StreamEntitiesRequest } from '@backstage/catalog-client'; import { ValidateEntityResponse } from '@backstage/catalog-client'; // @public @@ -70,6 +71,8 @@ export class InMemoryCatalogClient implements CatalogApi { // (undocumented) removeLocationById(_id: string): Promise; // (undocumented) + streamEntities(request?: StreamEntitiesRequest): AsyncIterable; + // (undocumented) validateEntity( _entity: Entity, _locationRef: string, diff --git a/packages/catalog-client/report.api.md b/packages/catalog-client/report.api.md index ebb0f5ac97..cda0d0c99d 100644 --- a/packages/catalog-client/report.api.md +++ b/packages/catalog-client/report.api.md @@ -88,6 +88,10 @@ export interface CatalogApi { id: string, options?: CatalogRequestOptions, ): Promise; + streamEntities( + request?: StreamEntitiesRequest, + options?: CatalogRequestOptions, + ): AsyncIterable; validateEntity( entity: Entity, locationRef: string, @@ -170,6 +174,10 @@ export class CatalogClient implements CatalogApi { id: string, options?: CatalogRequestOptions, ): Promise; + streamEntities( + request?: StreamEntitiesRequest, + options?: CatalogRequestOptions, + ): AsyncIterable; validateEntity( entity: Entity, locationRef: string, @@ -317,6 +325,12 @@ export type QueryEntitiesResponse = { }; }; +// @public +export type StreamEntitiesRequest = Omit< + QueryEntitiesRequest, + 'limit' | 'offset' +>; + // @public export type ValidateEntityResponse = | { diff --git a/packages/catalog-client/src/CatalogClient.test.ts b/packages/catalog-client/src/CatalogClient.test.ts index abafeb18a5..325f60cc13 100644 --- a/packages/catalog-client/src/CatalogClient.test.ts +++ b/packages/catalog-client/src/CatalogClient.test.ts @@ -540,6 +540,71 @@ describe('CatalogClient', () => { }); }); + describe('streamEntities', () => { + const defaultResponse: QueryEntitiesResponse = { + items: [ + { + apiVersion: '1', + kind: 'Component', + metadata: { + name: 'Test2', + namespace: 'test1', + }, + }, + { + apiVersion: '1', + kind: 'Component', + metadata: { + name: 'Test1', + namespace: 'test1', + }, + }, + ], + pageInfo: { + nextCursor: 'next', + prevCursor: 'prev', + }, + totalItems: 10, + }; + + beforeEach(() => { + server.use( + rest.get(`${mockBaseUrl}/entities/by-query`, (req, res, ctx) => { + const cursor = req.url.searchParams.get('cursor'); + if (cursor === 'next') { + return res(ctx.json({ items: [], pageInfo: {}, totalItems: 0 })); + } + return res(ctx.json(defaultResponse)); + }), + ); + }); + + it('should stream entities', async () => { + const stream = client.streamEntities({}, { token }); + const results: Entity[] = []; + for await (const entity of stream) { + results.push(entity); + } + expect(results).toEqual(defaultResponse.items); + }); + + it('should handle errors', async () => { + const mockedEndpoint = jest + .fn() + .mockImplementation((_req, res, ctx) => res(ctx.status(401))); + + server.use(rest.get(`${mockBaseUrl}/entities/by-query`, mockedEndpoint)); + + const stream = client.streamEntities({}, { token }); + await expect(async () => { + const results: Entity[] = []; + for await (const entity of stream) { + results.push(entity); + } + }).rejects.toThrow(/Request failed with 401 Unauthorized/); + }); + }); + describe('getEntityByRef', () => { const existingEntity: Entity = { apiVersion: 'v1', diff --git a/packages/catalog-client/src/CatalogClient.ts b/packages/catalog-client/src/CatalogClient.ts index 7d86eed95d..a412c790f6 100644 --- a/packages/catalog-client/src/CatalogClient.ts +++ b/packages/catalog-client/src/CatalogClient.ts @@ -40,6 +40,7 @@ import { Location, QueryEntitiesRequest, QueryEntitiesResponse, + StreamEntitiesRequest, ValidateEntityResponse, } from './types/api'; import { isQueryEntitiesInitialRequest, splitRefsIntoChunks } from './utils'; @@ -49,6 +50,9 @@ import type { AnalyzeLocationResponse, } from '@backstage/plugin-catalog-common'; +// Number of entities to return in a single streamEntities request +const STREAM_ENTITIES_LIMIT = 500; + /** * A frontend and backend compatible client for communicating with the Backstage * software catalog. @@ -456,6 +460,29 @@ export class CatalogClient implements CatalogApi { return response.json() as Promise; } + /** + * {@inheritdoc CatalogApi.streamEntities} + */ + async *streamEntities( + request?: StreamEntitiesRequest, + options?: CatalogRequestOptions, + ): AsyncIterable { + let cursor: string | undefined = undefined; + do { + const res = await this.queryEntities( + cursor + ? { ...request, cursor, limit: STREAM_ENTITIES_LIMIT } + : { ...request, limit: STREAM_ENTITIES_LIMIT }, + options, + ); + for (const entity of res.items) { + yield entity; + } + + cursor = res.pageInfo.nextCursor; + } while (cursor); + } + // // Private methods // diff --git a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts index c5d1f3abc8..5481c2ca3c 100644 --- a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts +++ b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts @@ -123,6 +123,16 @@ describe('InMemoryCatalogClient', () => { }); }); + it('streamEntities', async () => { + const client = new InMemoryCatalogClient({ entities }); + const stream = client.streamEntities(); + const results: Entity[] = []; + for await (const entity of stream) { + results.push(entity); + } + expect(results).toEqual(entities); + }); + it('getEntityAncestors', async () => { const client = new InMemoryCatalogClient({ entities }); await expect( diff --git a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts index 97663de922..a21709d798 100644 --- a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts +++ b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts @@ -32,6 +32,7 @@ import { Location, QueryEntitiesRequest, QueryEntitiesResponse, + StreamEntitiesRequest, ValidateEntityResponse, } from '@backstage/catalog-client'; import { @@ -279,6 +280,22 @@ export class InMemoryCatalogClient implements CatalogApi { throw new NotImplementedError('Method not implemented.'); } + async *streamEntities( + request?: StreamEntitiesRequest, + ): AsyncIterable { + let cursor: string | undefined = undefined; + do { + const res = await this.queryEntities( + cursor ? { ...request, cursor } : request, + ); + for (const entity of res.items) { + yield entity; + } + + cursor = res.pageInfo.nextCursor; + } while (cursor); + } + #createEntityRefMap() { return new Map(this.#entities.map(e => [stringifyEntityRef(e), e])); } diff --git a/packages/catalog-client/src/types/api.ts b/packages/catalog-client/src/types/api.ts index 28fe8bf276..9eae527a33 100644 --- a/packages/catalog-client/src/types/api.ts +++ b/packages/catalog-client/src/types/api.ts @@ -465,6 +465,16 @@ export type QueryEntitiesResponse = { }; }; +/** + * Stream entities request for {@link CatalogClient.streamEntities}. + * + * @public + */ +export type StreamEntitiesRequest = Omit< + QueryEntitiesRequest, + 'limit' | 'offset' +>; + /** * A client for interacting with the Backstage software catalog through its API. * @@ -692,4 +702,18 @@ export interface CatalogApi { location: AnalyzeLocationRequest, options?: CatalogRequestOptions, ): Promise; + + /** + * Asynchronously streams entities from the catalog. Uses `queryEntities` + * to fetch entities in batches, and yields them one by one. + * + * @public + * + * @param request - Request parameters + * @param options - Additional options + */ + streamEntities( + request?: StreamEntitiesRequest, + options?: CatalogRequestOptions, + ): AsyncIterable; } diff --git a/packages/catalog-client/src/types/index.ts b/packages/catalog-client/src/types/index.ts index 73b16f4554..5b483d9b29 100644 --- a/packages/catalog-client/src/types/index.ts +++ b/packages/catalog-client/src/types/index.ts @@ -38,5 +38,6 @@ export type { QueryEntitiesInitialRequest, QueryEntitiesRequest, QueryEntitiesResponse, + StreamEntitiesRequest, } from './api'; export { ENTITY_STATUS_CATALOG_PROCESSING_TYPE } from './status'; diff --git a/plugins/catalog-node/report-testUtils.api.md b/plugins/catalog-node/report-testUtils.api.md index 3fa7057a99..00f96ac347 100644 --- a/plugins/catalog-node/report-testUtils.api.md +++ b/plugins/catalog-node/report-testUtils.api.md @@ -27,6 +27,7 @@ import { QueryEntitiesRequest } from '@backstage/catalog-client'; import { QueryEntitiesResponse } from '@backstage/catalog-client'; import { ServiceFactory } from '@backstage/backend-plugin-api'; import { ServiceMock } from '@backstage/backend-test-utils'; +import { StreamEntitiesRequest } from '@backstage/catalog-client'; import { ValidateEntityResponse } from '@backstage/catalog-client'; // @public @@ -107,6 +108,11 @@ export interface CatalogServiceMock extends CatalogService, CatalogApi { options?: CatalogServiceRequestOptions | CatalogRequestOptions, ): Promise; // (undocumented) + streamEntities( + request?: StreamEntitiesRequest, + options?: CatalogServiceRequestOptions | CatalogRequestOptions, + ): AsyncIterable; + // (undocumented) validateEntity( entity: Entity, locationRef: string, diff --git a/plugins/catalog-node/report.api.md b/plugins/catalog-node/report.api.md index debafad5c4..5723666093 100644 --- a/plugins/catalog-node/report.api.md +++ b/plugins/catalog-node/report.api.md @@ -27,6 +27,7 @@ import { LocationSpec as LocationSpec_2 } from '@backstage/plugin-catalog-common import { QueryEntitiesRequest } from '@backstage/catalog-client'; import { QueryEntitiesResponse } from '@backstage/catalog-client'; import { ServiceRef } from '@backstage/backend-plugin-api'; +import { StreamEntitiesRequest } from '@backstage/catalog-client'; import { ValidateEntityResponse } from '@backstage/catalog-client'; // @public (undocumented) @@ -195,6 +196,11 @@ export interface CatalogService { options: CatalogServiceRequestOptions, ): Promise; // (undocumented) + streamEntities( + request: StreamEntitiesRequest | undefined, + options: CatalogServiceRequestOptions, + ): AsyncIterable; + // (undocumented) validateEntity( entity: Entity, locationRef: string, diff --git a/plugins/catalog-node/src/catalogService.ts b/plugins/catalog-node/src/catalogService.ts index 9478c6b48a..666a97274e 100644 --- a/plugins/catalog-node/src/catalogService.ts +++ b/plugins/catalog-node/src/catalogService.ts @@ -15,11 +15,11 @@ */ import { + AuthService, + BackstageCredentials, + coreServices, createServiceFactory, createServiceRef, - coreServices, - BackstageCredentials, - AuthService, } from '@backstage/backend-plugin-api'; import { AddLocationRequest, @@ -39,6 +39,7 @@ import { Location, QueryEntitiesRequest, QueryEntitiesResponse, + StreamEntitiesRequest, ValidateEntityResponse, } from '@backstage/catalog-client'; import { CompoundEntityRef, Entity } from '@backstage/catalog-model'; @@ -141,6 +142,11 @@ export interface CatalogService { location: AnalyzeLocationRequest, options: CatalogServiceRequestOptions, ): Promise; + + streamEntities( + request: StreamEntitiesRequest | undefined, + options: CatalogServiceRequestOptions, + ): AsyncIterable; } class DefaultCatalogService implements CatalogService { @@ -320,6 +326,16 @@ class DefaultCatalogService implements CatalogService { ); } + async *streamEntities( + request: StreamEntitiesRequest | undefined, + options: CatalogServiceRequestOptions, + ): AsyncIterable { + yield* this.#catalogApi.streamEntities( + request, + await this.#getOptions(options), + ); + } + async #getOptions( options: CatalogServiceRequestOptions, ): Promise { diff --git a/plugins/catalog-node/src/testUtils/catalogServiceMock.ts b/plugins/catalog-node/src/testUtils/catalogServiceMock.ts index 20e04e12b4..59c66c0b88 100644 --- a/plugins/catalog-node/src/testUtils/catalogServiceMock.ts +++ b/plugins/catalog-node/src/testUtils/catalogServiceMock.ts @@ -15,9 +15,9 @@ */ import { + createServiceFactory, ServiceFactory, ServiceRef, - createServiceFactory, } from '@backstage/backend-plugin-api'; import { InMemoryCatalogClient } from '@backstage/catalog-client/testUtils'; import { Entity } from '@backstage/catalog-model'; @@ -103,5 +103,6 @@ export namespace catalogServiceMock { getLocationByEntity: jest.fn(), validateEntity: jest.fn(), analyzeLocation: jest.fn(), + streamEntities: jest.fn(), })); } diff --git a/plugins/catalog-node/src/testUtils/types.ts b/plugins/catalog-node/src/testUtils/types.ts index 96380e9434..526527dcc9 100644 --- a/plugins/catalog-node/src/testUtils/types.ts +++ b/plugins/catalog-node/src/testUtils/types.ts @@ -31,6 +31,7 @@ import { Location, QueryEntitiesRequest, QueryEntitiesResponse, + StreamEntitiesRequest, ValidateEntityResponse, } from '@backstage/catalog-client'; import { CompoundEntityRef, Entity } from '@backstage/catalog-model'; @@ -135,4 +136,9 @@ export interface CatalogServiceMock extends CatalogService, CatalogApi { location: AnalyzeLocationRequest, options?: CatalogServiceRequestOptions | CatalogRequestOptions, ): Promise; + + streamEntities( + request?: StreamEntitiesRequest, + options?: CatalogServiceRequestOptions | CatalogRequestOptions, + ): AsyncIterable; } diff --git a/plugins/catalog-react/src/testUtils/catalogApiMock.ts b/plugins/catalog-react/src/testUtils/catalogApiMock.ts index 42c6b1004f..594b3a76be 100644 --- a/plugins/catalog-react/src/testUtils/catalogApiMock.ts +++ b/plugins/catalog-react/src/testUtils/catalogApiMock.ts @@ -102,5 +102,6 @@ export namespace catalogApiMock { getLocationByEntity: jest.fn(), validateEntity: jest.fn(), analyzeLocation: jest.fn(), + streamEntities: jest.fn(), })); } From 2f5162c4b783598e4ff7a1df494a1617edcc302a Mon Sep 17 00:00:00 2001 From: Hellgren Heikki Date: Fri, 29 Aug 2025 08:30:24 +0300 Subject: [PATCH 2/6] feat(catalog): batch size + mode config for entity streaming adds parameters to control the batch size and mode for the streamEntities method. with array mode, the entities are returned as batches while the default single mode returns entities one by one. Signed-off-by: Hellgren Heikki --- .../catalog-client/report-testUtils.api.md | 9 ++++++- packages/catalog-client/report.api.md | 19 +++++++++----- .../catalog-client/src/CatalogClient.test.ts | 13 ++++++++-- packages/catalog-client/src/CatalogClient.ts | 25 +++++++++++-------- .../testUtils/InMemoryCatalogClient.test.ts | 12 ++++++++- .../src/testUtils/InMemoryCatalogClient.ts | 17 +++++++++---- packages/catalog-client/src/types/api.ts | 13 ++++++++-- plugins/catalog-node/report-testUtils.api.md | 2 +- plugins/catalog-node/report.api.md | 2 +- plugins/catalog-node/src/catalogService.ts | 4 +-- plugins/catalog-node/src/testUtils/types.ts | 2 +- 11 files changed, 86 insertions(+), 32 deletions(-) diff --git a/packages/catalog-client/report-testUtils.api.md b/packages/catalog-client/report-testUtils.api.md index 26e79c9f96..cadf58aeaf 100644 --- a/packages/catalog-client/report-testUtils.api.md +++ b/packages/catalog-client/report-testUtils.api.md @@ -71,7 +71,14 @@ export class InMemoryCatalogClient implements CatalogApi { // (undocumented) removeLocationById(_id: string): Promise; // (undocumented) - streamEntities(request?: StreamEntitiesRequest): AsyncIterable; + streamEntities< + T extends StreamEntitiesRequest, + R = T extends { + mode: 'array'; + } & StreamEntitiesRequest + ? Entity[] + : Entity, + >(request?: T): AsyncIterable; // (undocumented) validateEntity( _entity: Entity, diff --git a/packages/catalog-client/report.api.md b/packages/catalog-client/report.api.md index cda0d0c99d..6000686fbd 100644 --- a/packages/catalog-client/report.api.md +++ b/packages/catalog-client/report.api.md @@ -91,7 +91,7 @@ export interface CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogRequestOptions, - ): AsyncIterable; + ): AsyncIterable; validateEntity( entity: Entity, locationRef: string, @@ -174,10 +174,14 @@ export class CatalogClient implements CatalogApi { id: string, options?: CatalogRequestOptions, ): Promise; - streamEntities( - request?: StreamEntitiesRequest, - options?: CatalogRequestOptions, - ): AsyncIterable; + streamEntities< + T extends StreamEntitiesRequest, + R = T extends { + mode: 'array'; + } & StreamEntitiesRequest + ? Entity[] + : Entity, + >(request?: T, options?: CatalogRequestOptions): AsyncIterable; validateEntity( entity: Entity, locationRef: string, @@ -329,7 +333,10 @@ export type QueryEntitiesResponse = { export type StreamEntitiesRequest = Omit< QueryEntitiesRequest, 'limit' | 'offset' ->; +> & { + batchSize?: number; + mode?: 'single' | 'array'; +}; // @public export type ValidateEntityResponse = diff --git a/packages/catalog-client/src/CatalogClient.test.ts b/packages/catalog-client/src/CatalogClient.test.ts index 325f60cc13..c5e2c8cc21 100644 --- a/packages/catalog-client/src/CatalogClient.test.ts +++ b/packages/catalog-client/src/CatalogClient.test.ts @@ -579,7 +579,7 @@ describe('CatalogClient', () => { ); }); - it('should stream entities', async () => { + it('should stream entities one by one', async () => { const stream = client.streamEntities({}, { token }); const results: Entity[] = []; for await (const entity of stream) { @@ -588,6 +588,15 @@ describe('CatalogClient', () => { expect(results).toEqual(defaultResponse.items); }); + it('should stream entities in batches', async () => { + const stream = client.streamEntities({ mode: 'array' }, { token }); + const results: Entity[] = []; + for await (const entityBatch of stream) { + results.push(...entityBatch); + } + expect(results).toEqual(defaultResponse.items); + }); + it('should handle errors', async () => { const mockedEndpoint = jest .fn() @@ -597,7 +606,7 @@ describe('CatalogClient', () => { const stream = client.streamEntities({}, { token }); await expect(async () => { - const results: Entity[] = []; + const results = []; for await (const entity of stream) { results.push(entity); } diff --git a/packages/catalog-client/src/CatalogClient.ts b/packages/catalog-client/src/CatalogClient.ts index a412c790f6..b090966807 100644 --- a/packages/catalog-client/src/CatalogClient.ts +++ b/packages/catalog-client/src/CatalogClient.ts @@ -51,7 +51,7 @@ import type { } from '@backstage/plugin-catalog-common'; // Number of entities to return in a single streamEntities request -const STREAM_ENTITIES_LIMIT = 500; +const DEFAULT_STREAM_ENTITIES_LIMIT = 500; /** * A frontend and backend compatible client for communicating with the Backstage @@ -463,20 +463,25 @@ export class CatalogClient implements CatalogApi { /** * {@inheritdoc CatalogApi.streamEntities} */ - async *streamEntities( - request?: StreamEntitiesRequest, - options?: CatalogRequestOptions, - ): AsyncIterable { + async *streamEntities< + T extends StreamEntitiesRequest, + R = T extends { mode: 'array' } & StreamEntitiesRequest ? Entity[] : Entity, + >(request?: T, options?: CatalogRequestOptions): AsyncIterable { let cursor: string | undefined = undefined; + const limit = request?.batchSize ?? DEFAULT_STREAM_ENTITIES_LIMIT; + const mode = request?.mode ?? 'single'; do { const res = await this.queryEntities( - cursor - ? { ...request, cursor, limit: STREAM_ENTITIES_LIMIT } - : { ...request, limit: STREAM_ENTITIES_LIMIT }, + cursor ? { ...request, cursor, limit } : { ...request, limit }, options, ); - for (const entity of res.items) { - yield entity; + + if (mode === 'single') { + for (const entity of res.items) { + yield entity as R; + } + } else { + yield res.items as R; } cursor = res.pageInfo.nextCursor; diff --git a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts index 5481c2ca3c..afc26346a7 100644 --- a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts +++ b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts @@ -123,7 +123,7 @@ describe('InMemoryCatalogClient', () => { }); }); - it('streamEntities', async () => { + it('streamEntities single', async () => { const client = new InMemoryCatalogClient({ entities }); const stream = client.streamEntities(); const results: Entity[] = []; @@ -133,6 +133,16 @@ describe('InMemoryCatalogClient', () => { expect(results).toEqual(entities); }); + it('streamEntities batch', async () => { + const client = new InMemoryCatalogClient({ entities }); + const stream = client.streamEntities({ mode: 'array' }); + const results: Entity[] = []; + for await (const entity of stream) { + results.push(...entity); + } + expect(results).toEqual(entities); + }); + it('getEntityAncestors', async () => { const client = new InMemoryCatalogClient({ entities }); await expect( diff --git a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts index a21709d798..fe23e7ceb1 100644 --- a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts +++ b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts @@ -280,16 +280,23 @@ export class InMemoryCatalogClient implements CatalogApi { throw new NotImplementedError('Method not implemented.'); } - async *streamEntities( - request?: StreamEntitiesRequest, - ): AsyncIterable { + async *streamEntities< + T extends StreamEntitiesRequest, + R = T extends { mode: 'array' } & StreamEntitiesRequest ? Entity[] : Entity, + >(request?: T): AsyncIterable { let cursor: string | undefined = undefined; + const mode = request?.mode ?? 'single'; do { const res = await this.queryEntities( cursor ? { ...request, cursor } : request, ); - for (const entity of res.items) { - yield entity; + + if (mode === 'single') { + for (const entity of res.items) { + yield entity as R; + } + } else { + yield res.items as R; } cursor = res.pageInfo.nextCursor; diff --git a/packages/catalog-client/src/types/api.ts b/packages/catalog-client/src/types/api.ts index 9eae527a33..fbf3d82a83 100644 --- a/packages/catalog-client/src/types/api.ts +++ b/packages/catalog-client/src/types/api.ts @@ -473,7 +473,16 @@ export type QueryEntitiesResponse = { export type StreamEntitiesRequest = Omit< QueryEntitiesRequest, 'limit' | 'offset' ->; +> & { + /** + * The number of entities to fetch in each batch. Defaults to 500. + */ + batchSize?: number; + /** + * The mode in which entities should be yielded. Defaults to 'single'. + */ + mode?: 'single' | 'array'; +}; /** * A client for interacting with the Backstage software catalog through its API. @@ -715,5 +724,5 @@ export interface CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogRequestOptions, - ): AsyncIterable; + ): AsyncIterable; } diff --git a/plugins/catalog-node/report-testUtils.api.md b/plugins/catalog-node/report-testUtils.api.md index 00f96ac347..4d0f7fdd48 100644 --- a/plugins/catalog-node/report-testUtils.api.md +++ b/plugins/catalog-node/report-testUtils.api.md @@ -111,7 +111,7 @@ export interface CatalogServiceMock extends CatalogService, CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogServiceRequestOptions | CatalogRequestOptions, - ): AsyncIterable; + ): AsyncIterable; // (undocumented) validateEntity( entity: Entity, diff --git a/plugins/catalog-node/report.api.md b/plugins/catalog-node/report.api.md index 5723666093..6e476f88f6 100644 --- a/plugins/catalog-node/report.api.md +++ b/plugins/catalog-node/report.api.md @@ -199,7 +199,7 @@ export interface CatalogService { streamEntities( request: StreamEntitiesRequest | undefined, options: CatalogServiceRequestOptions, - ): AsyncIterable; + ): AsyncIterable; // (undocumented) validateEntity( entity: Entity, diff --git a/plugins/catalog-node/src/catalogService.ts b/plugins/catalog-node/src/catalogService.ts index 666a97274e..d8509ddb20 100644 --- a/plugins/catalog-node/src/catalogService.ts +++ b/plugins/catalog-node/src/catalogService.ts @@ -146,7 +146,7 @@ export interface CatalogService { streamEntities( request: StreamEntitiesRequest | undefined, options: CatalogServiceRequestOptions, - ): AsyncIterable; + ): AsyncIterable; } class DefaultCatalogService implements CatalogService { @@ -329,7 +329,7 @@ class DefaultCatalogService implements CatalogService { async *streamEntities( request: StreamEntitiesRequest | undefined, options: CatalogServiceRequestOptions, - ): AsyncIterable { + ): AsyncIterable { yield* this.#catalogApi.streamEntities( request, await this.#getOptions(options), diff --git a/plugins/catalog-node/src/testUtils/types.ts b/plugins/catalog-node/src/testUtils/types.ts index 526527dcc9..fafdcb82fa 100644 --- a/plugins/catalog-node/src/testUtils/types.ts +++ b/plugins/catalog-node/src/testUtils/types.ts @@ -140,5 +140,5 @@ export interface CatalogServiceMock extends CatalogService, CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogServiceRequestOptions | CatalogRequestOptions, - ): AsyncIterable; + ): AsyncIterable; } From acb2fdb38fb2ed80364b0bb1ef837c89365ed38c Mon Sep 17 00:00:00 2001 From: Hellgren Heikki Date: Fri, 29 Aug 2025 14:11:05 +0300 Subject: [PATCH 3/6] feat(catalog): add possibility to stream entity pages adds a new method `streamEntityPages` that can be used to stream entity pages from catalog instead streaming entities one by one. this can be more efficient load wise but not usable for every use case. Signed-off-by: Hellgren Heikki --- .changeset/olive-moons-burn.md | 16 +++++++++- .../catalog-client/report-testUtils.api.md | 11 ++----- packages/catalog-client/report.api.md | 23 +++++++------ .../catalog-client/src/CatalogClient.test.ts | 14 ++++---- packages/catalog-client/src/CatalogClient.ts | 32 ++++++++++++------- .../testUtils/InMemoryCatalogClient.test.ts | 14 ++++---- .../src/testUtils/InMemoryCatalogClient.ts | 27 +++++++++------- packages/catalog-client/src/types/api.ts | 20 +++++++++--- plugins/catalog-node/report-testUtils.api.md | 7 +++- plugins/catalog-node/report.api.md | 7 +++- plugins/catalog-node/src/catalogService.ts | 19 +++++++++-- .../src/testUtils/catalogServiceMock.ts | 1 + plugins/catalog-node/src/testUtils/types.ts | 7 +++- .../src/testUtils/catalogApiMock.ts | 1 + 14 files changed, 132 insertions(+), 67 deletions(-) diff --git a/.changeset/olive-moons-burn.md b/.changeset/olive-moons-burn.md index feeff93d3d..652c8fff14 100644 --- a/.changeset/olive-moons-burn.md +++ b/.changeset/olive-moons-burn.md @@ -4,7 +4,7 @@ '@backstage/plugin-catalog-node': minor --- -Introduced new `streamEntities` async generator method for the catalog. +Introduced new `streamEntities` and `streamEntityPages` async generator methods for the catalog. Catalog API and Catalog Service now includes a `streamEntities` method that allows for streaming entities from the catalog. This method is designed to handle large datasets efficiently by processing entities in a stream rather than loading them @@ -19,3 +19,17 @@ for await (const entity of stream) { // Handle entity } ``` + +Additionally, a `streamEntityPages` method is available that streams entities in pages, allowing for batch processing of entities. +This method is more efficient than `streamEntities` when you can process entities in chunks. +Example usage: + +```ts +const pageStream = catalogClient.streamEntityPages( + { batchSize: 100 }, + { token }, +); +for await (const page of pageStream) { + // Handle page of entities +} +``` diff --git a/packages/catalog-client/report-testUtils.api.md b/packages/catalog-client/report-testUtils.api.md index cadf58aeaf..01bb94214b 100644 --- a/packages/catalog-client/report-testUtils.api.md +++ b/packages/catalog-client/report-testUtils.api.md @@ -71,14 +71,9 @@ export class InMemoryCatalogClient implements CatalogApi { // (undocumented) removeLocationById(_id: string): Promise; // (undocumented) - streamEntities< - T extends StreamEntitiesRequest, - R = T extends { - mode: 'array'; - } & StreamEntitiesRequest - ? Entity[] - : Entity, - >(request?: T): AsyncIterable; + streamEntities(request?: StreamEntitiesRequest): AsyncIterable; + // (undocumented) + streamEntityPages(request?: StreamEntitiesRequest): AsyncIterable; // (undocumented) validateEntity( _entity: Entity, diff --git a/packages/catalog-client/report.api.md b/packages/catalog-client/report.api.md index 6000686fbd..5bb36e16fc 100644 --- a/packages/catalog-client/report.api.md +++ b/packages/catalog-client/report.api.md @@ -91,7 +91,11 @@ export interface CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogRequestOptions, - ): AsyncIterable; + ): AsyncIterable; + streamEntityPages( + request?: StreamEntitiesRequest, + options?: CatalogRequestOptions, + ): AsyncIterable; validateEntity( entity: Entity, locationRef: string, @@ -174,14 +178,14 @@ export class CatalogClient implements CatalogApi { id: string, options?: CatalogRequestOptions, ): Promise; - streamEntities< - T extends StreamEntitiesRequest, - R = T extends { - mode: 'array'; - } & StreamEntitiesRequest - ? Entity[] - : Entity, - >(request?: T, options?: CatalogRequestOptions): AsyncIterable; + streamEntities( + request?: StreamEntitiesRequest, + options?: CatalogRequestOptions, + ): AsyncIterable; + streamEntityPages( + request?: StreamEntitiesRequest, + options?: CatalogRequestOptions, + ): AsyncIterable; validateEntity( entity: Entity, locationRef: string, @@ -335,7 +339,6 @@ export type StreamEntitiesRequest = Omit< 'limit' | 'offset' > & { batchSize?: number; - mode?: 'single' | 'array'; }; // @public diff --git a/packages/catalog-client/src/CatalogClient.test.ts b/packages/catalog-client/src/CatalogClient.test.ts index c5e2c8cc21..e330eefd33 100644 --- a/packages/catalog-client/src/CatalogClient.test.ts +++ b/packages/catalog-client/src/CatalogClient.test.ts @@ -579,7 +579,7 @@ describe('CatalogClient', () => { ); }); - it('should stream entities one by one', async () => { + it('should stream entities', async () => { const stream = client.streamEntities({}, { token }); const results: Entity[] = []; for await (const entity of stream) { @@ -588,13 +588,13 @@ describe('CatalogClient', () => { expect(results).toEqual(defaultResponse.items); }); - it('should stream entities in batches', async () => { - const stream = client.streamEntities({ mode: 'array' }, { token }); - const results: Entity[] = []; - for await (const entityBatch of stream) { - results.push(...entityBatch); + it('should stream entity pages', async () => { + const stream = client.streamEntityPages({}, { token }); + const results: Entity[][] = []; + for await (const entityPage of stream) { + results.push(entityPage); } - expect(results).toEqual(defaultResponse.items); + expect(results).toEqual([defaultResponse.items, []]); }); it('should handle errors', async () => { diff --git a/packages/catalog-client/src/CatalogClient.ts b/packages/catalog-client/src/CatalogClient.ts index b090966807..e83fd362da 100644 --- a/packages/catalog-client/src/CatalogClient.ts +++ b/packages/catalog-client/src/CatalogClient.ts @@ -463,26 +463,34 @@ export class CatalogClient implements CatalogApi { /** * {@inheritdoc CatalogApi.streamEntities} */ - async *streamEntities< - T extends StreamEntitiesRequest, - R = T extends { mode: 'array' } & StreamEntitiesRequest ? Entity[] : Entity, - >(request?: T, options?: CatalogRequestOptions): AsyncIterable { + async *streamEntities( + request?: StreamEntitiesRequest, + options?: CatalogRequestOptions, + ): AsyncIterable { + const pages = this.streamEntityPages(request, options); + for await (const page of pages) { + for (const entity of page) { + yield entity; + } + } + } + + /** + * {@inheritdoc CatalogApi.streamEntityPages} + */ + async *streamEntityPages( + request?: StreamEntitiesRequest, + options?: CatalogRequestOptions, + ): AsyncIterable { let cursor: string | undefined = undefined; const limit = request?.batchSize ?? DEFAULT_STREAM_ENTITIES_LIMIT; - const mode = request?.mode ?? 'single'; do { const res = await this.queryEntities( cursor ? { ...request, cursor, limit } : { ...request, limit }, options, ); - if (mode === 'single') { - for (const entity of res.items) { - yield entity as R; - } - } else { - yield res.items as R; - } + yield res.items; cursor = res.pageInfo.nextCursor; } while (cursor); diff --git a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts index afc26346a7..35f0a89248 100644 --- a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts +++ b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts @@ -123,7 +123,7 @@ describe('InMemoryCatalogClient', () => { }); }); - it('streamEntities single', async () => { + it('streamEntities', async () => { const client = new InMemoryCatalogClient({ entities }); const stream = client.streamEntities(); const results: Entity[] = []; @@ -133,14 +133,14 @@ describe('InMemoryCatalogClient', () => { expect(results).toEqual(entities); }); - it('streamEntities batch', async () => { + it('streamEntityPages', async () => { const client = new InMemoryCatalogClient({ entities }); - const stream = client.streamEntities({ mode: 'array' }); - const results: Entity[] = []; - for await (const entity of stream) { - results.push(...entity); + const stream = client.streamEntityPages(); + const results: Entity[][] = []; + for await (const page of stream) { + results.push(page); } - expect(results).toEqual(entities); + expect(results).toEqual([entities]); }); it('getEntityAncestors', async () => { diff --git a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts index fe23e7ceb1..c033ced387 100644 --- a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts +++ b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts @@ -280,24 +280,27 @@ export class InMemoryCatalogClient implements CatalogApi { throw new NotImplementedError('Method not implemented.'); } - async *streamEntities< - T extends StreamEntitiesRequest, - R = T extends { mode: 'array' } & StreamEntitiesRequest ? Entity[] : Entity, - >(request?: T): AsyncIterable { + async *streamEntities( + request?: StreamEntitiesRequest, + ): AsyncIterable { + const pages = this.streamEntityPages(request); + for await (const page of pages) { + for (const entity of page) { + yield entity; + } + } + } + + async *streamEntityPages( + request?: StreamEntitiesRequest, + ): AsyncIterable { let cursor: string | undefined = undefined; - const mode = request?.mode ?? 'single'; do { const res = await this.queryEntities( cursor ? { ...request, cursor } : request, ); - if (mode === 'single') { - for (const entity of res.items) { - yield entity as R; - } - } else { - yield res.items as R; - } + yield res.items; cursor = res.pageInfo.nextCursor; } while (cursor); diff --git a/packages/catalog-client/src/types/api.ts b/packages/catalog-client/src/types/api.ts index fbf3d82a83..56872c60c8 100644 --- a/packages/catalog-client/src/types/api.ts +++ b/packages/catalog-client/src/types/api.ts @@ -478,10 +478,6 @@ export type StreamEntitiesRequest = Omit< * The number of entities to fetch in each batch. Defaults to 500. */ batchSize?: number; - /** - * The mode in which entities should be yielded. Defaults to 'single'. - */ - mode?: 'single' | 'array'; }; /** @@ -724,5 +720,19 @@ export interface CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogRequestOptions, - ): AsyncIterable; + ): AsyncIterable; + + /** + * Asynchronously streams entity pages from the catalog. Uses `queryEntities` + * to fetch entities in batches, and yields them one page at a time. + * + * @public + * + * @param request - Request parameters + * @param options - Additional options + */ + streamEntityPages( + request?: StreamEntitiesRequest, + options?: CatalogRequestOptions, + ): AsyncIterable; } diff --git a/plugins/catalog-node/report-testUtils.api.md b/plugins/catalog-node/report-testUtils.api.md index 4d0f7fdd48..e06fb741d1 100644 --- a/plugins/catalog-node/report-testUtils.api.md +++ b/plugins/catalog-node/report-testUtils.api.md @@ -111,7 +111,12 @@ export interface CatalogServiceMock extends CatalogService, CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogServiceRequestOptions | CatalogRequestOptions, - ): AsyncIterable; + ): AsyncIterable; + // (undocumented) + streamEntityPages( + request?: StreamEntitiesRequest, + options?: CatalogServiceRequestOptions | CatalogRequestOptions, + ): AsyncIterable; // (undocumented) validateEntity( entity: Entity, diff --git a/plugins/catalog-node/report.api.md b/plugins/catalog-node/report.api.md index 6e476f88f6..9173f89e29 100644 --- a/plugins/catalog-node/report.api.md +++ b/plugins/catalog-node/report.api.md @@ -199,7 +199,12 @@ export interface CatalogService { streamEntities( request: StreamEntitiesRequest | undefined, options: CatalogServiceRequestOptions, - ): AsyncIterable; + ): AsyncIterable; + // (undocumented) + streamEntityPages( + request: StreamEntitiesRequest | undefined, + options: CatalogServiceRequestOptions, + ): AsyncIterable; // (undocumented) validateEntity( entity: Entity, diff --git a/plugins/catalog-node/src/catalogService.ts b/plugins/catalog-node/src/catalogService.ts index d8509ddb20..aa6a69a69b 100644 --- a/plugins/catalog-node/src/catalogService.ts +++ b/plugins/catalog-node/src/catalogService.ts @@ -146,7 +146,12 @@ export interface CatalogService { streamEntities( request: StreamEntitiesRequest | undefined, options: CatalogServiceRequestOptions, - ): AsyncIterable; + ): AsyncIterable; + + streamEntityPages( + request: StreamEntitiesRequest | undefined, + options: CatalogServiceRequestOptions, + ): AsyncIterable; } class DefaultCatalogService implements CatalogService { @@ -329,13 +334,23 @@ class DefaultCatalogService implements CatalogService { async *streamEntities( request: StreamEntitiesRequest | undefined, options: CatalogServiceRequestOptions, - ): AsyncIterable { + ): AsyncIterable { yield* this.#catalogApi.streamEntities( request, await this.#getOptions(options), ); } + async *streamEntityPages( + request: StreamEntitiesRequest | undefined, + options: CatalogServiceRequestOptions, + ): AsyncIterable { + yield* this.#catalogApi.streamEntityPages( + request, + await this.#getOptions(options), + ); + } + async #getOptions( options: CatalogServiceRequestOptions, ): Promise { diff --git a/plugins/catalog-node/src/testUtils/catalogServiceMock.ts b/plugins/catalog-node/src/testUtils/catalogServiceMock.ts index 59c66c0b88..78962dc484 100644 --- a/plugins/catalog-node/src/testUtils/catalogServiceMock.ts +++ b/plugins/catalog-node/src/testUtils/catalogServiceMock.ts @@ -104,5 +104,6 @@ export namespace catalogServiceMock { validateEntity: jest.fn(), analyzeLocation: jest.fn(), streamEntities: jest.fn(), + streamEntityPages: jest.fn(), })); } diff --git a/plugins/catalog-node/src/testUtils/types.ts b/plugins/catalog-node/src/testUtils/types.ts index fafdcb82fa..c8ada67dc5 100644 --- a/plugins/catalog-node/src/testUtils/types.ts +++ b/plugins/catalog-node/src/testUtils/types.ts @@ -140,5 +140,10 @@ export interface CatalogServiceMock extends CatalogService, CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogServiceRequestOptions | CatalogRequestOptions, - ): AsyncIterable; + ): AsyncIterable; + + streamEntityPages( + request?: StreamEntitiesRequest, + options?: CatalogServiceRequestOptions | CatalogRequestOptions, + ): AsyncIterable; } diff --git a/plugins/catalog-react/src/testUtils/catalogApiMock.ts b/plugins/catalog-react/src/testUtils/catalogApiMock.ts index 594b3a76be..a37d2669ad 100644 --- a/plugins/catalog-react/src/testUtils/catalogApiMock.ts +++ b/plugins/catalog-react/src/testUtils/catalogApiMock.ts @@ -103,5 +103,6 @@ export namespace catalogApiMock { validateEntity: jest.fn(), analyzeLocation: jest.fn(), streamEntities: jest.fn(), + streamEntityPages: jest.fn(), })); } From 713a875c5b1eaef757085c4e2c29d9b281eb3104 Mon Sep 17 00:00:00 2001 From: Hellgren Heikki Date: Mon, 1 Sep 2025 15:02:55 +0300 Subject: [PATCH 4/6] fix: review comments of naming Signed-off-by: Hellgren Heikki --- packages/catalog-client/report.api.md | 4 ++-- packages/catalog-client/src/CatalogClient.ts | 6 ++---- packages/catalog-client/src/constants.ts | 17 +++++++++++++++++ .../src/testUtils/InMemoryCatalogClient.ts | 4 +++- packages/catalog-client/src/types/api.ts | 6 +++--- 5 files changed, 27 insertions(+), 10 deletions(-) create mode 100644 packages/catalog-client/src/constants.ts diff --git a/packages/catalog-client/report.api.md b/packages/catalog-client/report.api.md index 5bb36e16fc..4175cf3a4b 100644 --- a/packages/catalog-client/report.api.md +++ b/packages/catalog-client/report.api.md @@ -335,10 +335,10 @@ export type QueryEntitiesResponse = { // @public export type StreamEntitiesRequest = Omit< - QueryEntitiesRequest, + QueryEntitiesInitialRequest, 'limit' | 'offset' > & { - batchSize?: number; + pageSize?: number; }; // @public diff --git a/packages/catalog-client/src/CatalogClient.ts b/packages/catalog-client/src/CatalogClient.ts index e83fd362da..c777c48940 100644 --- a/packages/catalog-client/src/CatalogClient.ts +++ b/packages/catalog-client/src/CatalogClient.ts @@ -49,9 +49,7 @@ import type { AnalyzeLocationRequest, AnalyzeLocationResponse, } from '@backstage/plugin-catalog-common'; - -// Number of entities to return in a single streamEntities request -const DEFAULT_STREAM_ENTITIES_LIMIT = 500; +import { DEFAULT_STREAM_ENTITIES_LIMIT } from './constants.ts'; /** * A frontend and backend compatible client for communicating with the Backstage @@ -483,7 +481,7 @@ export class CatalogClient implements CatalogApi { options?: CatalogRequestOptions, ): AsyncIterable { let cursor: string | undefined = undefined; - const limit = request?.batchSize ?? DEFAULT_STREAM_ENTITIES_LIMIT; + const limit = request?.pageSize ?? DEFAULT_STREAM_ENTITIES_LIMIT; do { const res = await this.queryEntities( cursor ? { ...request, cursor, limit } : { ...request, limit }, diff --git a/packages/catalog-client/src/constants.ts b/packages/catalog-client/src/constants.ts new file mode 100644 index 0000000000..dd8d0530df --- /dev/null +++ b/packages/catalog-client/src/constants.ts @@ -0,0 +1,17 @@ +/* + * Copyright 2025 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +export const DEFAULT_STREAM_ENTITIES_LIMIT = 500; diff --git a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts index c033ced387..ef125d3c3a 100644 --- a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts +++ b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts @@ -49,6 +49,7 @@ import type { AnalyzeLocationRequest, AnalyzeLocationResponse, } from '@backstage/plugin-catalog-common'; +import { DEFAULT_STREAM_ENTITIES_LIMIT } from '../constants.ts'; function buildEntitySearch(entity: Entity) { const rows = traverse(entity); @@ -295,9 +296,10 @@ export class InMemoryCatalogClient implements CatalogApi { request?: StreamEntitiesRequest, ): AsyncIterable { let cursor: string | undefined = undefined; + const limit = request?.pageSize ?? DEFAULT_STREAM_ENTITIES_LIMIT; do { const res = await this.queryEntities( - cursor ? { ...request, cursor } : request, + cursor ? { ...request, limit, cursor } : { ...request, limit }, ); yield res.items; diff --git a/packages/catalog-client/src/types/api.ts b/packages/catalog-client/src/types/api.ts index 56872c60c8..6a83b95d6e 100644 --- a/packages/catalog-client/src/types/api.ts +++ b/packages/catalog-client/src/types/api.ts @@ -471,13 +471,13 @@ export type QueryEntitiesResponse = { * @public */ export type StreamEntitiesRequest = Omit< - QueryEntitiesRequest, + QueryEntitiesInitialRequest, 'limit' | 'offset' > & { /** - * The number of entities to fetch in each batch. Defaults to 500. + * The number of entities to fetch in each page. Defaults to 500. */ - batchSize?: number; + pageSize?: number; }; /** From a87ab00712c579dcad06573e7d5bf6f8694d2226 Mon Sep 17 00:00:00 2001 From: Heikki Hellgren Date: Mon, 1 Sep 2025 21:16:29 +0300 Subject: [PATCH 5/6] fix: changeset pageSize parameter Signed-off-by: Heikki Hellgren --- .changeset/olive-moons-burn.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/olive-moons-burn.md b/.changeset/olive-moons-burn.md index 652c8fff14..42c488e637 100644 --- a/.changeset/olive-moons-burn.md +++ b/.changeset/olive-moons-burn.md @@ -26,7 +26,7 @@ Example usage: ```ts const pageStream = catalogClient.streamEntityPages( - { batchSize: 100 }, + { pageSize: 100 }, { token }, ); for await (const page of pageStream) { From 9118c56520db40477561fc567a3bc45ae30f2923 Mon Sep 17 00:00:00 2001 From: Hellgren Heikki Date: Wed, 3 Sep 2025 08:48:10 +0300 Subject: [PATCH 6/6] fix(catalog): remove streamEntityPages function and make the streamEntities work as the pages function did earlier Signed-off-by: Hellgren Heikki --- .changeset/olive-moons-burn.md | 21 +++++-------------- .../catalog-client/report-testUtils.api.md | 4 +--- packages/catalog-client/report.api.md | 8 ------- .../catalog-client/src/CatalogClient.test.ts | 9 -------- packages/catalog-client/src/CatalogClient.ts | 15 ------------- .../testUtils/InMemoryCatalogClient.test.ts | 10 --------- .../src/testUtils/InMemoryCatalogClient.ts | 11 ---------- packages/catalog-client/src/types/api.ts | 16 +------------- plugins/catalog-node/report-testUtils.api.md | 5 ----- plugins/catalog-node/report.api.md | 5 ----- plugins/catalog-node/src/catalogService.ts | 17 +-------------- .../src/testUtils/catalogServiceMock.ts | 1 - plugins/catalog-node/src/testUtils/types.ts | 5 ----- 13 files changed, 8 insertions(+), 119 deletions(-) diff --git a/.changeset/olive-moons-burn.md b/.changeset/olive-moons-burn.md index 42c488e637..625d97c5a7 100644 --- a/.changeset/olive-moons-burn.md +++ b/.changeset/olive-moons-burn.md @@ -4,7 +4,7 @@ '@backstage/plugin-catalog-node': minor --- -Introduced new `streamEntities` and `streamEntityPages` async generator methods for the catalog. +Introduced new `streamEntities` async generator method for the catalog. Catalog API and Catalog Service now includes a `streamEntities` method that allows for streaming entities from the catalog. This method is designed to handle large datasets efficiently by processing entities in a stream rather than loading them @@ -14,22 +14,11 @@ or fetch all entities at once. Example usage: ```ts -const stream = catalogClient.streamEntities({}, { token }); -for await (const entity of stream) { - // Handle entity -} -``` - -Additionally, a `streamEntityPages` method is available that streams entities in pages, allowing for batch processing of entities. -This method is more efficient than `streamEntities` when you can process entities in chunks. -Example usage: - -```ts -const pageStream = catalogClient.streamEntityPages( - { pageSize: 100 }, - { token }, -); +const pageStream = catalogClient.streamEntities({ pageSize: 100 }, { token }); for await (const page of pageStream) { // Handle page of entities + for (const entity of page) { + console.log(entity); + } } ``` diff --git a/packages/catalog-client/report-testUtils.api.md b/packages/catalog-client/report-testUtils.api.md index 01bb94214b..b96c4f01ca 100644 --- a/packages/catalog-client/report-testUtils.api.md +++ b/packages/catalog-client/report-testUtils.api.md @@ -71,9 +71,7 @@ export class InMemoryCatalogClient implements CatalogApi { // (undocumented) removeLocationById(_id: string): Promise; // (undocumented) - streamEntities(request?: StreamEntitiesRequest): AsyncIterable; - // (undocumented) - streamEntityPages(request?: StreamEntitiesRequest): AsyncIterable; + streamEntities(request?: StreamEntitiesRequest): AsyncIterable; // (undocumented) validateEntity( _entity: Entity, diff --git a/packages/catalog-client/report.api.md b/packages/catalog-client/report.api.md index 4175cf3a4b..452d788ba9 100644 --- a/packages/catalog-client/report.api.md +++ b/packages/catalog-client/report.api.md @@ -91,10 +91,6 @@ export interface CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogRequestOptions, - ): AsyncIterable; - streamEntityPages( - request?: StreamEntitiesRequest, - options?: CatalogRequestOptions, ): AsyncIterable; validateEntity( entity: Entity, @@ -181,10 +177,6 @@ export class CatalogClient implements CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogRequestOptions, - ): AsyncIterable; - streamEntityPages( - request?: StreamEntitiesRequest, - options?: CatalogRequestOptions, ): AsyncIterable; validateEntity( entity: Entity, diff --git a/packages/catalog-client/src/CatalogClient.test.ts b/packages/catalog-client/src/CatalogClient.test.ts index e330eefd33..8307f77cdd 100644 --- a/packages/catalog-client/src/CatalogClient.test.ts +++ b/packages/catalog-client/src/CatalogClient.test.ts @@ -581,15 +581,6 @@ describe('CatalogClient', () => { it('should stream entities', async () => { const stream = client.streamEntities({}, { token }); - const results: Entity[] = []; - for await (const entity of stream) { - results.push(entity); - } - expect(results).toEqual(defaultResponse.items); - }); - - it('should stream entity pages', async () => { - const stream = client.streamEntityPages({}, { token }); const results: Entity[][] = []; for await (const entityPage of stream) { results.push(entityPage); diff --git a/packages/catalog-client/src/CatalogClient.ts b/packages/catalog-client/src/CatalogClient.ts index c777c48940..07e6ad2946 100644 --- a/packages/catalog-client/src/CatalogClient.ts +++ b/packages/catalog-client/src/CatalogClient.ts @@ -464,21 +464,6 @@ export class CatalogClient implements CatalogApi { async *streamEntities( request?: StreamEntitiesRequest, options?: CatalogRequestOptions, - ): AsyncIterable { - const pages = this.streamEntityPages(request, options); - for await (const page of pages) { - for (const entity of page) { - yield entity; - } - } - } - - /** - * {@inheritdoc CatalogApi.streamEntityPages} - */ - async *streamEntityPages( - request?: StreamEntitiesRequest, - options?: CatalogRequestOptions, ): AsyncIterable { let cursor: string | undefined = undefined; const limit = request?.pageSize ?? DEFAULT_STREAM_ENTITIES_LIMIT; diff --git a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts index 35f0a89248..dbe3390747 100644 --- a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts +++ b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.test.ts @@ -126,16 +126,6 @@ describe('InMemoryCatalogClient', () => { it('streamEntities', async () => { const client = new InMemoryCatalogClient({ entities }); const stream = client.streamEntities(); - const results: Entity[] = []; - for await (const entity of stream) { - results.push(entity); - } - expect(results).toEqual(entities); - }); - - it('streamEntityPages', async () => { - const client = new InMemoryCatalogClient({ entities }); - const stream = client.streamEntityPages(); const results: Entity[][] = []; for await (const page of stream) { results.push(page); diff --git a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts index ef125d3c3a..a9db99ab77 100644 --- a/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts +++ b/packages/catalog-client/src/testUtils/InMemoryCatalogClient.ts @@ -283,17 +283,6 @@ export class InMemoryCatalogClient implements CatalogApi { async *streamEntities( request?: StreamEntitiesRequest, - ): AsyncIterable { - const pages = this.streamEntityPages(request); - for await (const page of pages) { - for (const entity of page) { - yield entity; - } - } - } - - async *streamEntityPages( - request?: StreamEntitiesRequest, ): AsyncIterable { let cursor: string | undefined = undefined; const limit = request?.pageSize ?? DEFAULT_STREAM_ENTITIES_LIMIT; diff --git a/packages/catalog-client/src/types/api.ts b/packages/catalog-client/src/types/api.ts index 6a83b95d6e..014baba613 100644 --- a/packages/catalog-client/src/types/api.ts +++ b/packages/catalog-client/src/types/api.ts @@ -710,7 +710,7 @@ export interface CatalogApi { /** * Asynchronously streams entities from the catalog. Uses `queryEntities` - * to fetch entities in batches, and yields them one by one. + * to fetch entities in batches, and yields them one page at a time. * * @public * @@ -720,19 +720,5 @@ export interface CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogRequestOptions, - ): AsyncIterable; - - /** - * Asynchronously streams entity pages from the catalog. Uses `queryEntities` - * to fetch entities in batches, and yields them one page at a time. - * - * @public - * - * @param request - Request parameters - * @param options - Additional options - */ - streamEntityPages( - request?: StreamEntitiesRequest, - options?: CatalogRequestOptions, ): AsyncIterable; } diff --git a/plugins/catalog-node/report-testUtils.api.md b/plugins/catalog-node/report-testUtils.api.md index e06fb741d1..d687ff387b 100644 --- a/plugins/catalog-node/report-testUtils.api.md +++ b/plugins/catalog-node/report-testUtils.api.md @@ -111,11 +111,6 @@ export interface CatalogServiceMock extends CatalogService, CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogServiceRequestOptions | CatalogRequestOptions, - ): AsyncIterable; - // (undocumented) - streamEntityPages( - request?: StreamEntitiesRequest, - options?: CatalogServiceRequestOptions | CatalogRequestOptions, ): AsyncIterable; // (undocumented) validateEntity( diff --git a/plugins/catalog-node/report.api.md b/plugins/catalog-node/report.api.md index 9173f89e29..25ab9145ef 100644 --- a/plugins/catalog-node/report.api.md +++ b/plugins/catalog-node/report.api.md @@ -199,11 +199,6 @@ export interface CatalogService { streamEntities( request: StreamEntitiesRequest | undefined, options: CatalogServiceRequestOptions, - ): AsyncIterable; - // (undocumented) - streamEntityPages( - request: StreamEntitiesRequest | undefined, - options: CatalogServiceRequestOptions, ): AsyncIterable; // (undocumented) validateEntity( diff --git a/plugins/catalog-node/src/catalogService.ts b/plugins/catalog-node/src/catalogService.ts index aa6a69a69b..a6ecedd5c5 100644 --- a/plugins/catalog-node/src/catalogService.ts +++ b/plugins/catalog-node/src/catalogService.ts @@ -146,11 +146,6 @@ export interface CatalogService { streamEntities( request: StreamEntitiesRequest | undefined, options: CatalogServiceRequestOptions, - ): AsyncIterable; - - streamEntityPages( - request: StreamEntitiesRequest | undefined, - options: CatalogServiceRequestOptions, ): AsyncIterable; } @@ -334,18 +329,8 @@ class DefaultCatalogService implements CatalogService { async *streamEntities( request: StreamEntitiesRequest | undefined, options: CatalogServiceRequestOptions, - ): AsyncIterable { - yield* this.#catalogApi.streamEntities( - request, - await this.#getOptions(options), - ); - } - - async *streamEntityPages( - request: StreamEntitiesRequest | undefined, - options: CatalogServiceRequestOptions, ): AsyncIterable { - yield* this.#catalogApi.streamEntityPages( + yield* this.#catalogApi.streamEntities( request, await this.#getOptions(options), ); diff --git a/plugins/catalog-node/src/testUtils/catalogServiceMock.ts b/plugins/catalog-node/src/testUtils/catalogServiceMock.ts index 78962dc484..59c66c0b88 100644 --- a/plugins/catalog-node/src/testUtils/catalogServiceMock.ts +++ b/plugins/catalog-node/src/testUtils/catalogServiceMock.ts @@ -104,6 +104,5 @@ export namespace catalogServiceMock { validateEntity: jest.fn(), analyzeLocation: jest.fn(), streamEntities: jest.fn(), - streamEntityPages: jest.fn(), })); } diff --git a/plugins/catalog-node/src/testUtils/types.ts b/plugins/catalog-node/src/testUtils/types.ts index c8ada67dc5..a5bf125cfd 100644 --- a/plugins/catalog-node/src/testUtils/types.ts +++ b/plugins/catalog-node/src/testUtils/types.ts @@ -140,10 +140,5 @@ export interface CatalogServiceMock extends CatalogService, CatalogApi { streamEntities( request?: StreamEntitiesRequest, options?: CatalogServiceRequestOptions | CatalogRequestOptions, - ): AsyncIterable; - - streamEntityPages( - request?: StreamEntitiesRequest, - options?: CatalogServiceRequestOptions | CatalogRequestOptions, ): AsyncIterable; }