Add an initial implementation of realtime subscriptions. #45
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
andrewchambers
force-pushed
the
realtimesubs
branch
7 times, most recently
from
03c0ac3 to
d448eb5
Compare
| @@ -386,3 +386,25 @@ def sql_query(self, stmt, params: typing.List): | |||
| for page in self.stub.SQLQuery(request): | |||
| check_proto_stat(page.stat) | |||
| yield page.SQLQueryRow | |||
|
|
|||
| @error_handler | |||
| def subscribe(self, update_queue): | |||
There was a problem hiding this comment.
Can we add type hinting for the update_queue parameter?
There was a problem hiding this comment.
Yeah, I think I will do a pass of type hinting + docstrings + tests. The queue is slightly awkward because of how grpc handles bidirectional streams.
| import threading | ||
| import weakref | ||
|
|
||
| from bytewax.inputs import DynamicInput, StatelessSource |
There was a problem hiding this comment.
Not important yet, but will need to update the pyproject.toml optional dependencies to include bytewax
This implementation also includes an example bytewax adapter
which can be used to process inserts with bytewax.
An example bytewax flow is shown below:
```
import uuid
import btrdb
from btrdb.experimental.bytewax_connectors import InsertSubscription
from bytewax.dataflow import Dataflow
from bytewax.connectors.stdio import StdOutput
def selector(db):
# Selector can be anything that returns a list of uuids.
rows = db.query('select uuid from streams')
uuids = [uuid.UUID(row['uuid']) for row in rows]
return uuids
flow = Dataflow()
flow.input("realtime_sub", InsertSubscription(selector, selector_refresh_interval=30))
flow.output("print_output", StdOutput())
```
andrewchambers
force-pushed
the
realtimesubs
branch
from
d448eb5 to
7993e26
Compare
jleifnf
reviewed
Sep 22, 2023
| self._background_worker.start() | ||
| self._read_worker.start() | ||
|
|
||
| def next(self): |
There was a problem hiding this comment.
I attempted to use the bytewax flow example, and it seems in bytewax=='0.17.1' there is no next it's either next_batch or next_awake
There was a problem hiding this comment.
Yeah, this was written against v0.16, needs to be reworked slightly for 0.17.
…57) * update setuptools parameters for strict standards starting 2023-Oct-30 * remove invalid url for homepage of the project * remove invalid url for homepage of the project * update the docs url in pyproject.toml
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This implementation also includes an example bytewax adapter which can be used to process realtime inserts with bytewax.
An example bytewax flow is shown below: