observeQuery improvement

[maxludovicohofer]

Is this related to a new or existing framework?

Angular, React, React Native, Vue, Web Components, Next.js, New framework

Is this related to a new or existing API?

DataStore

Is this related to another service?

No response

Describe the feature you'd like to request

observeQuery should apply the provided filter predicate on observe also, to check for subsequent results.
Also, it would be nice if observeQuery had paging support.

Describe the solution you'd like

// Should check if isValid is still true when query items change
DataStore.observeQuery(
    type,
    ({ isValid }) => isValid("eq", true),
    {
        // Should support page and limit
        page,
        limit,
        sort
    }
);

Describe alternatives you've considered

Getting items through observeQuery first; when synced, observing the model with observe.
If there is a mutation, running query (with the same filter predicate) only on the mutated item, and removing it or adding it or changing it.

For paging, I use query and then observe the model. If there is a mutation, I rerun the query on the model.

Additional context

No response

Is this something that you'd be interested in working on?

• 👋 I may be able to implement this feature request
• ⚠️ This feature might incur a breaking change

[nickarocho]

Hi @maxludovicohofer, thanks for your question.

Observe query does support dynamic sorting, filtering, and limiting the amount of records received in snapshots, so I'd like to fully understand your individual use case better to provide some personalized recommendations with the current implementation of the observeQuery API.

Though I'm sure you've found our documentation on this API already, for those who aren't fully familiar - here is the Observe Query documentation which describes common use cases like sorting, filtering, and configuring limits for the sync page size.

---

You mentioned that observeQuery:

Should check if isValid is still true when query items change

Could you please describe a little more about what isValid would be referring to in your app and what you mean by "when query items change"? Is isValid a field on a given model? How might the query items change?

re: the limit (i.e. max records or items to sync within each returned snapshot)

you can configure this by adjusting the syncPageSize via:

DataStore.configure({
  syncPageSize: 100, // default is 1000
});

re: pagination

Could you also describe why you would need pagination while using observeQuery? The API behaves just as observe, which also doesn't take a pagination argument, but if you could provide a little more insight into your specific use case we might be able to help you achieve what you are looking to do in your app.

Overall, it would be helpful to know:

• the schema of the model you want to use with observeQuery
• the framework you are using for your app (React, Vue, Angular, etc.)
• the specific user experience you expect (a brief description of your use case and your "ideal scenario" as a developer)
• some code snippets documenting how you are using DataStore to perform queries, mutations, observing models, etc.

Thanks!

[maxludovicohofer]

Use case
I want to query data to display in a table/list. When data is changed (data is added/updated/deleted), the list should react to this change and rerender the table/list with the correct data.
Overall, I would like observeQuery to behave more like "query" and less like "observe".

My problem:
The predicate i pass to observeQuery is not checked when data is updated.
Let's say that the model of the data has a boolean property "isValid", and I want to query only the elements where isValid is true.

type Example @model {
  id: ID!
  isValid: Boolean!
}

DataStore.observeQuery(
    Example,
    ({ isValid }) => isValid("eq", true)
);

Until isSync is true, the results will be correct.
But, if I update an element of the list to have "isValid" equal to false, this change will not trigger a reaction in the observeQuery. Meaning, the row corresponding to the element of the list will remain rendered, while it should not be rendered anymore (render is triggered when elements change).

Limit
I would like to have an SQL-like limit, similar to the "query" function. It looks like "sync page size" only fragments the results into smaller chunks.

Pagination
I would like to have a pagination similar to the "query" function.

Real world example
Here is how I implemented my use case, using datastore in react.

// THIS IS HOW I USE OBSERVE QUERY
export function storeList<T extends Entity>(
  type: PersistentModelConstructor<T>,
  onResult: (entities: T[]) => void,
  condition?: ProducerModelPredicate<T> | typeof Predicates.ALL,
  paginationProducer?: ObserveQueryOptions<T>,
  includeDeleted?: boolean
) {
  const subscriptionWrapper = {} as {
    subscription: ReturnType<ReturnType<typeof storeObserve>["subscribe"]>;
  };

  let async = true;

  const querySubscription = DataStore.observeQuery(
    type,
    includeDeleted ? condition : evaluateDeleted(type, condition),
    paginationProducer
  ).subscribe(({ items, isSynced }) => {
    if (isSynced || items.length !== 0) onResult(items);

    if (isSynced) {
      async = false;

      subscriptionWrapper.subscription = storeObserve(type).subscribe(
        async () => {
          const changedItems = await storeQuery(
            type,
            condition,
            paginationProducer,
            includeDeleted
          );

          if (changedItems) onResult(changedItems);
        }
      );

      querySubscription.unsubscribe();
    }
  });

  if (async) subscriptionWrapper.subscription = querySubscription;

  return subscriptionWrapper;
}

// THIS IS HOW I HANDLE THE RERENDER OF THE LIST
/**
 * Subscribes to a model and returns its values.
 * @param list The list operation to subscribe to
 * @returns Null if the query is still loading, else the results
 */
export function useList<T>(
  list: (
    subscribe: (result: T[]) => void | Promise<void>
  ) => Promise<ReturnType<typeof storeList> | undefined> | undefined
) {
  const [listResult, setListResult] = useState<T[] | null>(null);

  let observe: ReturnType<typeof storeList> | undefined;

  useAsync(
    {
      condition: !observe,
      action: async (isMounted) =>
        list((entities) => {
          if (isMounted()) {
            //? If entities are added or deleted, or updated, update list
            if (
              listResult?.length !== entities.length ||
              entities.some(
                (entity, index) =>
                  JSON.stringify(listResult?.[index]) !== JSON.stringify(entity)
              )
            ) {
              setListResult(entities);
            }
          }
        }),
      cleanup: () => observe?.subscription.unsubscribe(),
    },
    (subscription) => (observe = subscription)
  );

  return listResult;
}

// THE FOLLOWING FUNCTIONS ARE ADDED FOR COMPLETENESS, BUT ARE NOT THE FOCUS OF THIS EXAMPLE
function evaluateDeleted<T extends Entity>(
  type: PersistentModelConstructor<T>,
  condition?: ProducerModelPredicate<T> | typeof Predicates.ALL
) {
  return deleteable[getEntityName(type)]
    ? ((({ deletedAt }) => {
        const predicate = deletedAt("eq" as any, null as any); // deletedAt is a property of all my models

        return condition === Predicates.ALL
          ? predicate
          : condition?.(predicate) ?? predicate;
      }) as typeof condition)
    : condition;
}

/**
 * Loads data asynchronously, without the boilerplate.
 * @param fetch The object to construct the function
 * @param load Callback executed after fetch has retrieved the data successfully
 */
export function useAsync<T>(
  fetch:
    | {
        action: (isMounted: () => boolean) => T | Promise<T>;
        condition?: boolean;
        cleanup?: () => void;
      }
    | undefined,
  load?: (data: T) => void
) {
  useEffect(() => {
    let isMounted = true;

    async function loadData() {
      let data = undefined;
      let isError = false;

      try {
        data = await fetch!.action(() => isMounted);
      } catch (error) {
        log(error);
        isError = true;
      }

      if (!isError && isMounted) load?.(data!);
    }

    if (fetch) {
      if (fetch.condition !== false) loadData();

      if (fetch.cleanup) {
        return () => {
          isMounted = false;
          fetch.cleanup!();
        };
      }
    }
  });
}

[rnrnstar2]

I would also like to add this function.
If you update the data displayed in the table, the query will not catch the change and will remain in the previous display.
I want to use the page function and sort function, so I would appreciate it if you could add it to observeQuery.

[chrisbonifacio]

Hi @maxludovicohofer @rnrnstar2 👋 we've released a potential fix a while back. If you're still experiencing this issue, could you please update to the latest version of aws-amplify and/or @aws-amplify/datastore and verify whether or not this issue persists for you?

[rnrnstar2]

@chrisbonifacio

Thank you for your support!
I upgraded to the latest version and tried.
"@aws-amplify/datastore": "^3.14.0",
But I got a typo and page and limit didn't work either.

[erinleigh90]

@rnrnstar2, I would suggest taking a look at the datastore documentation on observeQuery, where it specifies that you can use predicates and sorting as in query, but does not include pagination. Specifically see this quote from the bottom of the page

The default configuration for observeQuery has a default syncPageSize of 1,000 and a default maxRecordsToSync of 10,000. These values can be customized by configuring their values manually in DataStore.configure."

ObserveQuery is meant to provide a snapshot of records as they are being synced and notify when it is done syncing. It does this by paginating through records in the server 1k (default) at a time per the pageSyncSize, up to the value of maxRecordsToSync (10k default).
We suggest slicing a subset of results from the snapshot and paginating locally through those results.

As far as the original issue brought up by @maxludovicohofer, this should have been resolved in later releases of Amplify, so I will go ahead and close this ticket. If you are still having issues with observeQuery I would suggest opening your own ticket with detailed instructions for reproduction. Thank you!

[kolodi]

It is indeed confusing.
Pagination parameter ProducerPaginationInput<T> for both query and queryObserve functions is the same, leading to unexpected results. Documentation is not clear at all about this fact.

Here are some useful snippets for different use cases

You can, of course, use queryObserve for simple cases when you do not need pagination, usually when you expect resulting list to be very small:

import { DataStore, PersistentModel, PersistentModelConstructor, ProducerPaginationInput, RecursiveModelPredicateExtender } from "aws-amplify/datastore";
import { useEffect, useState } from "react";

type ObserveQueryProps<T extends PersistentModel> = {
    modelConstructor: PersistentModelConstructor<T>
    criteria?: RecursiveModelPredicateExtender<T>
    paginationProducer?: ProducerPaginationInput<T>
}

/**
 * Observed query for list items. Note: Pagination is not working properly here. For pagination use useQueryList
 */
export default function useDataStoreListObserved<T extends PersistentModel>({
    modelConstructor,
    criteria,
    paginationProducer,
}: ObserveQueryProps<T>) {

    const [items, setItems] = useState<T[]>([])

    useEffect(() => {
        const sub = DataStore
            .observeQuery(modelConstructor, criteria, paginationProducer)
            .subscribe(({ items }) => {
                setItems(items)
            })

        return () => sub.unsubscribe()
    }, [modelConstructor, criteria, paginationProducer])

    return items
}

If you need a proper pagination I generally use query and manually triggering "refetch" when needed instead of relying on subscriptions events which usually triggers multiple times per update causing undesired rerenders.

import { useQuery } from "@tanstack/react-query";
import { DataStore, PersistentModel, PersistentModelConstructor, ProducerPaginationInput, RecursiveModelPredicateExtender } from "aws-amplify/datastore";
import { useState } from "react";

type ObserveQueryProps<T extends PersistentModel> = {
    modelConstructor: PersistentModelConstructor<T>
    criteria?: RecursiveModelPredicateExtender<T>
    paginationProducer?: ProducerPaginationInput<T>
    criteriaDeps?: any[]
}

/**
 * Query items list from DataStore with working pagination. 
 * Use triggerQuery function to manually trigger refetch.
 * Criteria is a function and it cannot be used as a dependance to trigger refetch,
 * therefore use a separate criteriaDeps, most often the same deps used to recreate criteria
 */
export default function useListQuery<T extends PersistentModel>({
    modelConstructor,
    criteria,
    paginationProducer,
    criteriaDeps = [],
}: ObserveQueryProps<T>) {

    const [v, setV] = useState(0)

    function triggerQuery() {
        setV(p => p + 1)
    }

    const { data: items, isLoading, error } = useQuery({
        queryKey: [modelConstructor, paginationProducer, updateVersion, ...criteriaDeps],
        queryFn: async () => DataStore.query(modelConstructor, criteria, paginationProducer),
        initialData: [],
    })

    return {
        items,
        isLoading,
        error,
        triggerQuery,
    }
}

Or you can combine both query and queryObserve, and inside subscription just trigger query refetch. I usually do this when working with a single item using observe:

import { useEffect, useState } from "react"
import { useQuery } from "@tanstack/react-query"
import {
    DataStore,
    IdentifierFieldOrIdentifierObject,
    PersistentModel,
    PersistentModelConstructor,
    PersistentModelMetaData,
} from "aws-amplify/datastore"

export default function useDataStoreItemObserved<T extends PersistentModel>(
    itemType: PersistentModelConstructor<T>,
    id: IdentifierFieldOrIdentifierObject<T, PersistentModelMetaData<T>>,
) {
    const [v, setV] = useState(0)
    const query = useQuery({
        queryKey: [itemType, id, v],
        queryFn: () => DataStore.query(itemType, id),
    })

    useEffect(() => {
        const sub = DataStore
            .observe(itemType, id as string)
            .subscribe(() => {
                setV(vv => vv + 1)
            })

        return () => sub.unsubscribe()
    }, [itemType, id])

    return query
}

Note useQuery here is just to solve annoying async state management issues in react and not for actual network requests.