A composable job query that handles pagination internally.
Start one with Zizq.query/1, narrow it, then run it with anything
from Enum or Stream — a query is enumerable, so pages are
fetched as they are needed and no further:
Zizq.query(MyApp.Zizq)
|> Zizq.Query.where(queue: "emails", status: [:ready])
|> Enum.take(10)That stops after the first page, because Enum.take/2 stops asking.
The same query run to completion walks every page:
Zizq.query(MyApp.Zizq)
|> Zizq.Query.where(queue: "emails")
|> Enum.each(&IO.inspect/1)Building a query performs no request. Nothing is sent until it is
enumerated, counted, or run through update_all/2 or
delete_all/1.
Narrowing
where/2 takes the filters Zizq.Filter documents, and can be
called repeatedly — later calls win per key, so a query can be
built up in pieces:
base = Zizq.query(MyApp.Zizq) |> Zizq.Query.where(queue: "emails")
base |> Zizq.Query.where(status: :ready)
base |> Zizq.Query.where(status: :dead)One where/2 rather than a by_queue, by_status, by_type and
so on for each field: the filters are already a keyword list, and
the server takes them as one set.
Counting
Enum.count/1 asks the server to count rather than fetching every
page, so it costs one request whatever the total:
Zizq.query(MyApp.Zizq) |> Zizq.Query.where(queue: "emails") |> Enum.count()Limits and page size
limit/2 caps how many jobs come back in total; in_pages_of/2
sets how many are fetched per request. They are independent — the
first is what you want, the second is how eagerly it is fetched.
Working in batches
update_all/2 and delete_all/1 normally send one request and let
the server do the work from the filters, which is what you want for
anything of ordinary size.
Give the query a limit/2 and/or an in_pages_of/2 and they switch
to working a page at a time instead, acting on each page by id:
Zizq.query(MyApp.Zizq)
|> Zizq.Query.where(queue: "emails", status: :dead)
|> Zizq.Query.in_pages_of(1_000)
|> Zizq.Query.delete_all()Ten million jobs then become a run of bounded requests rather than one enormous one. Each page's ids are sent with the original filters rather than instead of them, so a job that stopped matching between being listed and being acted on is left alone.
The count returned is the total across every batch either way.
Summary
Functions
How many jobs the query would yield, without fetching them.
Delete every job the query matches, and return how many.
Set how many jobs are fetched per request.
Cap how many jobs the query returns in total.
Return jobs oldest first (:asc) or newest first (:desc).
The pages themselves, as a Stream of Zizq.JobPage.
Change every job the query matches, and return how many.
Narrow the query. See Zizq.Filter for what can be given.
Types
@type t() :: %Zizq.Query{ client: atom(), filters: keyword(), limit: pos_integer() | nil, order: :asc | :desc | nil, page_size: pos_integer() | nil }
Functions
@spec count(t()) :: non_neg_integer()
How many jobs the query would yield, without fetching them.
One request whatever the total. Enum.count/1 on a query calls
this.
A limit/2 caps this as it caps everything else, so counting and
enumerating agree — Enum.count(query) and
length(Enum.to_list(query)) are the same number, reached by
different routes.
@spec delete_all(t()) :: non_neg_integer()
Delete every job the query matches, and return how many.
One request by default. See "Working in batches".
@spec in_pages_of(t(), pos_integer()) :: t()
Set how many jobs are fetched per request.
Independent of limit/2: this is how eagerly the query pages, not
how much it returns.
@spec limit(t(), pos_integer()) :: t()
Cap how many jobs the query returns in total.
Return jobs oldest first (:asc) or newest first (:desc).
@spec pages(t()) :: Enumerable.t()
The pages themselves, as a Stream of Zizq.JobPage.
For when a page at a time is the useful unit — acknowledging in batches, say — rather than a flat run of jobs.
@spec update_all( t(), keyword() ) :: non_neg_integer()
Change every job the query matches, and return how many.
Takes the options Zizq.update_job/3 takes.
Zizq.query(MyApp.Zizq)
|> Zizq.Query.where(queue: "emails", status: :scheduled)
|> Zizq.Query.update_all(ready_at: nil)One request, whatever the total — the server does the work from the filters. See "Working in batches" for when that is not what you want.
Narrow the query. See Zizq.Filter for what can be given.
Merges with what is already there, so later calls win per key.