Signal-first server state primitives for Angular, built on RxJS Observable.

Observable<T> in, Signal<QueryState<T>> out. You keep RxJS at the transport edge and manipulate typed state everywhere else — no manual subscriptions, no accidental double-fetches, no untyped loading flags.

📚 Documentation & live playground: https://angular-query.klheb.fr/

Visit the documentation website for the complete guides, API reference, interactive examples and live playground.

The two secondary entry points are opt-in, so the core stays small.

npm install @klheb/angular-query

Requires Angular 21+. Works with or without Zone.js (provideZonelessChangeDetection()).

useQuery creates a hook factory. The returned function takes Angular signals as arguments; if any argument is null/undefined, the query is disabled until every argument is available.

import { Injectable, inject } from '@angular/core';
import { useQuery, QueryRefetchOnInputChange } from '@klheb/angular-query';

@Injectable({ providedIn: 'root' })
export class UserStore {
  private readonly api = inject(UserApi);

  readonly user = useQuery((id: number) => ({
    key: ['user', id] as const,
    query: () => this.api.getUser(id),
    staleTime: 30_000,
  }));

  // Stable key, reactive input: a search box that refetches as the term changes.
  readonly search = useQuery((term: string) => ({
    key: ['users', 'search'] as const,
    query: () => this.api.search(term),
    refetchOnInputChange: QueryRefetchOnInputChange.ALWAYS,
  }));
}

Config:

• key: cache key for the server resource.
• query: function returning Observable<TData>.
• staleTime: how long cached data stays fresh (default 30s).
• gcTime: how long an inactive entry is kept before removal (default 5min).
• refetchOnMount: STALE (default), ALWAYS, or NEVER.
• refetchOnInputChange: NEVER (default) or ALWAYS, for stable-key queries whose input signals change (search / autocomplete).

Returned API:

• state: signal of IDLE | LOADING | SUCCESS | ERROR.
• refetch(): refetches; a running request is cancelled and replaced by the latest one.
• cancel(): cancels the running request and resets to IDLE.

import { useMutation, MutationConcurrency, MutationResetOnNoSubscribers } from '@klheb/angular-query';

readonly createUser = useMutation(() => ({
  key: ['create-user'] as const,
  mutation: (payload: CreateUserDto) => this.api.createUser(payload),
  resetOnNoSubscribers: MutationResetOnNoSubscribers.SUCCESS,
  invalidateKeys: () => [['users'] as const],
}));

// A "regenerate" action: relaunching cancels the in-flight run and starts a new one.
readonly regenerate = useMutation(() => ({
  key: ['regenerate'] as const,
  mutation: (prompt: string) => this.api.generate(prompt),
  concurrency: MutationConcurrency.REPLACE,
}));

Config:

• key: cache key for this mutation state.
• mutation: function returning Observable<TData>.
• concurrency: DEDUPE (default — a running mutation ignores new calls, so a double click never replays a POST) or REPLACE (cancel the running mutation and start the new one).
• resetOnNoSubscribers: NEVER (default), SUCCESS, ERROR, or SETTLED.
• gcTime: how long an inactive entry stays cached (default 5min). A running mutation is never collected — collection waits for it to settle.
• invalidateKeys: query keys invalidated after a successful mutation.

Behavior options (staleTime, gcTime, refetchOnMount, concurrency, resetOnNoSubscribers…) are static: read once when the config first resolves, constant afterwards. Only the key and the transport react to argument changes.

Why mutations do take concurrency. Unlike a query, mutate() is imperative, so the intent when it is called again mid-flight is genuinely ambiguous — ignore the duplicate, or replace it? That is a real choice, so it is exposed. (Same principle, opposite conclusion from queries.)

Returned API:

• state: signal of the mutation state.
• mutate(payload, callbacks?): starts the mutation. Callbacks: onSuccess, onError, onComplete.
• reset(): cancels the active mutation and resets to IDLE.

import {
  useInvalidateKey,
  useInvalidatePrefix,
  useInvalidateWhere,
} from '@klheb/angular-query';

const invalidateKey = useInvalidateKey();
const invalidatePrefix = useInvalidatePrefix();
const invalidateWhere = useInvalidateWhere();

invalidateKey(['users']);
invalidatePrefix(['users']); // ['users'], ['users', 1], ['users', { page: 2 }] …
invalidateWhere((query) => query.key[0] === 'user');

Active queries refetch immediately; inactive queries are marked stale and refetch the next time a component subscribes. A failed fetch also marks its entry stale, so errors are never served as fresh data.

For optimistic updates, setQueryData writes into an existing entry and marks it fresh:

const setQueryData = useSetQueryData();

setQueryData<User[]>(['users'], (users) => [...(users ?? []), optimisticUser]);

useStreamQuery keeps an accumulated state while an Observable emits many events (SSE, AI streams, WebSocket). Each event is folded into the state through a reducer.

import { useStreamQuery } from '@klheb/angular-query';

readonly chat = useStreamQuery((roomId: number) => ({
  key: ['room', roomId, 'chat'] as const,
  initialData: () => ({ messages: [] as string[] }),
  call: (prompt: string) => this.api.streamChat(roomId, prompt),
  reducer: (state, event) => ({ messages: [...(state?.messages ?? []), event.delta] }),
}));

Concurrency modes: QUEUE (default — one call at a time, order preserved), PARALLEL, IGNORE, REPLACE. The state also exposes an endReason (COMPLETED, CANCELLED, REPLACED, ERROR, IDLE). API: call, cancel, reset, setData.

import { OpenAiBaseReducer, createOpenAiStreamState } from '@klheb/angular-query/openai';

readonly assistant = useStreamQuery((userId: number) => ({
  key: ['assistant', userId] as const,
  initialData: createOpenAiStreamState,
  call: (prompt: string) => this.api.streamResponses(prompt),
  reducer: OpenAiBaseReducer,
}));

OpenAiBaseReducer normalizes lifecycle, text deltas, refusals, reasoning, tool calls, MCP, code interpreter and image events. Extend it with createOpenAiReducer({ initialState, handlers, afterBaseReducer }).

An optional, opinionated layer to declare a domain's keys and endpoints in one typed place. Auth belongs in HttpInterceptors, not here:

import { createQueryStore, query, mutation } from '@klheb/angular-query/store';

export const AppStore = createQueryStore({ api: UserApi });

export const UsersStore = AppStore.createSubStore({
  keys: { detail: (id: number) => ['user', id] as const },
  endpoints: ({ api, keys }) => ({
    detail: query((id: number) => ({ key: keys.detail(id), queryFn: () => api.getUser(id) })),
    update: mutation((id: number) => ({
      key: [...keys.detail(id), 'update'] as const,
      mutationFn: (name: string) => api.updateUser(id, name),
      invalidateKeys: () => [keys.detail(id)],
    })),
  }),
});

protected readonly STATUS = STATUS;
protected readonly userId = signal(1);
protected readonly userQuery = this.userStore.user(this.userId);

@let state = user.state();

@switch (state.status) {
  @case (STATUS.LOADING) { <app-loader /> }
  @case (STATUS.ERROR)   { <p>{{ state.error.message }}</p> }
  @case (STATUS.SUCCESS) { <h1>{{ state.data.name }}</h1> }
}

MIT