Query Network State Service

An implementation of the QueryNetworkStateService will provide an object-oriented wrapper for the gRPC library, with the ability to retrieve information about the state of the network. This is done with the following 3 steps:

1. Create callbacks which will be called when gRPC messages are received from a client.
2. Register the callbacks with an instance of the QueryNetworkStateService.
3. Add your QueryNetworkStateService instance as a gRPC service.

Creating callbacks

Each supported message in QueryNetworkStateService requires a callback, which will be triggered when any client messages are received.

on_get_current_states

The on_get_current_states callback is triggered for each request for the current states between two date/times, and should yield the batches of current state events that occurred between those date/times (inclusive)

from datetime import datetime
from typing import AsyncGenerator

from zepben.ewb import CurrentStateEventBatch


async def on_get_current_states(from_datetime: datetime, to_datetime: datetime) -> AsyncGenerator[CurrentStateEventBatch, None]:
    batch_id = 1
    events = []
    # build the list of events for this batch
    yield CurrentStateEventBatch(batch_id, events)

Each CurrentStateEventBatch you yield must carry the batch_id that will be sent to the client, as that is the ID the client will report its status against.

on_current_states_status

The on_current_states_status callback is triggered for each status response sent by the client. You should expect to receive one of these for every batch returned from on_get_current_states.

from zepben.ewb import SetCurrentStatesStatus


def on_current_states_status(event_status: SetCurrentStatesStatus):
    # event_status.batch_id is the ID of the batch returned from on_get_current_states.
    print(f"Client handled batch {event_status.batch_id}: {type(event_status).__name__}")

on_processing_error

The on_processing_error callback is triggered for any error raised in your on_current_states_status callback, or if a SetCurrentStatesResponse received from the client is for an unknown event status. It is always called with a message describing the problem, and the exception that caused it - which will be None when the client sent an unknown status rather than your callback raising.

from typing import Optional


def on_processing_error(message: str, exception: Optional[Exception]):
    # Log the message and the exception, which may be None.
    print(f"{message}: {exception}")

Registering callbacks

Registering the callbacks with the service is as simple as passing them into the QueryNetworkStateService constructor.

service = QueryNetworkStateService(on_get_current_states, on_current_states_status, on_processing_error)

Registering the service

For the above code to have any effect, you need to register the service with a gRPC server. Once this has been done, you should start to receive callbacks for each request sent from a gRPC client.

import grpc
from zepben.protobuf.ns.network_state_pb2_grpc import add_QueryNetworkStateServiceServicer_to_server

from zepben.ewb import QueryNetworkStateService


async def start_server(service: QueryNetworkStateService, host: str = 'localhost:50051'):
    server = grpc.aio.server()
    server.add_insecure_port(host)

    add_QueryNetworkStateServiceServicer_to_server(service, server)

    await server.start()

Putting it all together

Putting each of the steps above together, you can build the scaffold of a working application

from datetime import datetime
from typing import AsyncGenerator, Optional

import grpc
from zepben.protobuf.ns.network_state_pb2_grpc import add_QueryNetworkStateServiceServicer_to_server

from zepben.ewb import CurrentStateEventBatch, QueryNetworkStateService, SetCurrentStatesStatus

class QueryNetworkStateServiceImpl:

    def __init__(self):
        self.service = QueryNetworkStateService(self.on_get_current_states, self.on_current_states_status, self.on_processing_error)

    async def on_get_current_states(self, from_datetime: datetime, to_datetime: datetime) -> AsyncGenerator[CurrentStateEventBatch, None]:
        batch_id = 1
        events = []
        # build the list of events for this batch
        yield CurrentStateEventBatch(batch_id, events)

    def on_current_states_status(self, event_status: SetCurrentStatesStatus):
        print(f"Client handled batch {event_status.batch_id}: {type(event_status).__name__}")

    def on_processing_error(self, message: str, exception: Optional[Exception]):
        print(f"{message}: {exception}")

async def main():
    server = grpc.aio.server()
    host = 'localhost:50051'
    server.add_insecure_port(host)

    service = QueryNetworkStateServiceImpl().service
    add_QueryNetworkStateServiceServicer_to_server(service, server)

    await server.start()