Metadata-Version: 2.4
Name: event-reduce
Version: 7.5.1
Summary: Update live query results from database change events without re-running the query.
License-Expression: MIT
Project-URL: Homepage, https://github.com/pubkey/event-reduce
Project-URL: Repository, https://github.com/pubkey/event-reduce
Keywords: live-query,reactive,realtime,database,bdd
Requires-Python: >=3.10
Description-Content-Type: text/markdown

# event-reduce (Python)

Python implementation of the [EventReduce algorithm](https://github.com/pubkey/event-reduce).
It takes the previous results of a query and a change event of the database
and calculates the new results **in memory, without running the query over the database again**.

It uses the BDD that the [JavaScript implementation](../javascript/) created,
so both always decide on the same action. It needs Python 3.10 or newer and has no dependencies.

## Installation

```sh
pip install event-reduce
```

## Usage

```python
from event_reduce import (
    ActionName,
    ChangeEvent,
    StateResolveFunctionInput,
    calculate_action_name,
    dict_query_params,
    run_action,
)


def compare(a, b):
    # must never return 0 for two different documents, so sort by the primary key last
    if a["age"] != b["age"]:
        return -1 if a["age"] < b["age"] else 1
    return -1 if a["id"] < b["id"] else 1


# SELECT * FROM users WHERE age > 18 ORDER BY age, id LIMIT 10
query_params = dict_query_params(
    primary_key="id",
    sort_fields=["age", "id"],
    query_matcher=lambda doc: doc["age"] > 18,
    sort_comparator=compare,
    limit=10,
)

results = [{"id": "a", "age": 20}, {"id": "b", "age": 30}]

# a change event from the change stream of your database
change_event = ChangeEvent.insert("c", {"id": "c", "age": 25})

action = calculate_action_name(StateResolveFunctionInput(query_params, change_event, results))
if action == ActionName.RUN_FULL_QUERY_AGAIN:
    results = run_query_on_database()  # EventReduce could not optimize this event
else:
    run_action(action, query_params, change_event, results)
    # -> [a, c, b]
```

`dict_query_params()` is for dict documents, like the `primaryKey` and `sortFields` of the JavaScript implementation.
The sort fields can be dot-separated paths like `"address.city"`.
For other documents, create the `QueryParams` with a `get_document_id` function
and a `sort_fields_changed(previous, doc)` function that returns true if a sort field changed.

### Key document map

For big result sets, pass a dict from the id to the document as `key_document_map`
to both `StateResolveFunctionInput` and `run_action()`.
It is used to check whether a document is in the results without scanning them,
and `run_action()` keeps it in sync with the results.

### Differences to the JavaScript implementation

- Change events are created with `ChangeEvent.insert()`, `ChangeEvent.update()` and `ChangeEvent.delete()`.
- The names are snake_case, `ActionName` and `StateName` are enums whose values are the JavaScript names.
- `run_action()` raises an `EventReduceError` for `RUN_FULL_QUERY_AGAIN` and `UNKNOWN_ACTION`.
  `ALWAYS_WRONG` is only used by the truth-table generator and raises like `UNKNOWN_ACTION`.

## Result window buffer

For a query with a limit or a skip, EventReduce returns `RUN_FULL_QUERY_AGAIN` when a row has to move into the result window that is not in the previous results:
the row after the window (when a row leaves a full window) or the last skipped row (when a row is added in front of the window).
The result window buffer avoids most of these full queries: it keeps `limit_buffer_size` rows after the window and `skip_buffer_size` skipped rows before it in memory.
It works like the [result window buffer of the JavaScript implementation](../docs/result-window-buffer.md),
which explains how it works, how to choose the buffer sizes and has measurements.

```python
from event_reduce import ResultWindowBuffer, get_recommended_result_window_buffer_size

# recommended size: about log2(limit), at most 30
size = get_recommended_result_window_buffer_size(query_params.limit)
buffer = ResultWindowBuffer(query_params, limit_buffer_size=size, skip_buffer_size=size)
# run the query with the skip and limit of the buffer
buffer.reset(run_query(buffer.query()))
# the rows of the original skip and limit
results = buffer.results()

# on new change events
result = buffer.apply_change_events(change_events)
if result.run_full_query_again:
    buffer.reset(run_query(buffer.query()))
    results = buffer.results()
elif result.changed:
    results = result.results
```

`query()` returns the `skip` and `limit` to run the query with, `limit` is None if all rows after the skip are needed.
`math.inf` as a buffer size keeps all rows on that side, like `Infinity` in JavaScript.

## Development

```sh
pip install pytest ruff mypy
pytest
ruff check . && ruff format --check . && mypy src
```

The BDD is generated by the truth-table generator of the JavaScript implementation,
which also writes `src/event_reduce/bdd_generated.py`.
The tests run the [shared test suite](../test-suite/), the same tests as all other implementations.

## Who uses EventReduce

[RxDB](https://rxdb.info/), the local-first database, uses EventReduce for its live queries since [version 9.0.0](https://rxdb.info/releases/9.0.0.html), where it is enabled by default.
When an app subscribes to a query, RxDB keeps the results up to date by running each write event through EventReduce,
and only queries the storage again when EventReduce returns `RUN_FULL_QUERY_AGAIN`.
