semvec/pypi-packages huggingface dataset:
datastreamagent 0.1.2
A package which enables the creation and management of subscriptions to a data stream. When any data is manipulated, the diff of the data is sent to subscribers of that data stream
Requires Python: >=3.8,<4.0

#Data Stream Manager

A package which enables the creation and management of subscriptions to a data stream. When any data is manipulated, the diff of the data is sent to subscribers of that data stream

Supports Python 3.8+

Example usage is given below:

To handle subscriptions, your handler should call the appropriate subscription methods. The example below demonstrates this:

```python
from awspytools.DynamoDBDataStore import DynamoDBDataStore
from datastreamagent.dynamodb_stream_agent import DynamoDBDataStreamAgent

store = DynamoDBDataStore(table_name='table')

dynamodb_stream_manager = DynamoDBDataStreamAgent(store=store)


def on_subscribe(message, connection_id):
    dynamodb_stream_manager.subscribe(
        client_id=connection_id,
        identifier=message.get("identifier"),
        session_id=message.get("session_id")
    )


def on_unsubscribe(message, connection_id):
    dynamodb_stream_manager.unsubscribe(
        client_id=connection_id,
        identifier=message.get("identifier"),
        session_id=message.get("session_id")
    )
```

When an update is made, the updated record needs to be passed into the data stream manager, this updated record is generated by the data store that you choose. The example below demonstrates the use of a DynamoDB store where the updated record is generated from a dynamodb stream:

```python
from awspytools.DynamoDBDataStore import DynamoDBDataStore
from dynamodb_json import json_util
from datastreamagent.dynamodb_stream_agent import DynamoDBDataStreamAgent
from datastreamagent.models import SequenceRecord

store = DynamoDBDataStore(table_name='table')

dynamodb_stream_manager = DynamoDBDataStreamManager(store=store)


class Websocket:
    pass

    @staticmethod
    def send_data_to_connection(connection_id, data):
        pass


def send_message_to_subscribers(record, client_id):
    Websocket.send_data_to_connection(connection_id=client_id, data=record)


def process_item(item):
    dynamodb_document = item['dynamodb'].get('NewImage')

    document = json_util.loads(dynamodb_document)

    document.pop("PK", None)
    document.pop("SK", None)

    record = SequenceRecord(document)

    dynamodb_stream_manager.send_update_to_subscriptions(
        record=record,
        sending_client=send_message_to_subscribers
    )


def dynamodb_stream_trigger(event, context):
    items = event['Records']
    for item in items:
        process_item(item)
```