Skip to content

Get info

Get Info about a persistent subscription.

get_info(stub, group_name, stream_name=None, **kwargs)

Gets info about a persistent subscription.

Source code in src/eventstore_grpc/persistent/get_info.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def get_info(
    stub: persistent_pb2_grpc.PersistentSubscriptionsStub,
    group_name: str,
    stream_name: Optional[str] = None,
    **kwargs,
) -> persistent_pb2.GetInfoResp:
    """Gets info about a persistent subscription."""
    options = persistent_pb2.GetInfoReq.Options(group_name=group_name)
    if stream_name is None:  # $all
        options.all.CopyFrom(shared_pb2.Empty())
    else:
        options.stream_identifier.CopyFrom(
            shared_pb2.StreamIdentifier(stream_name=stream_name.encode())
        )
    request = persistent_pb2.GetInfoReq(options=options)
    response = stub.GetInfo(request, **kwargs)
    return response