fastapi with grpc server (using Depends, generator)

Question

I am coding a FastAPI code that acts as a client to communicate with a gRPC server on another server. I wanted to simplify the process of creating and closing the gRPC client object similar to how we can use dependency injection (Depends) to handle the session object of FastAPI's database. But first, I would like to confirm if I'm thinking correctly. Is it the right approach to inject the dependency on the stub that communicates with the gRPC server? Since I need to perform gRPC communication in around 5 APIs, I thought about creating a generator for connecting and disconnecting the channel.

here is my code. I got error AttributeError: 'CalculationStub' object has no attribute 'get_result' but I've defined a function called get_result inside the GRPCClient class, so I'm not sure what the problem is.

class GRPCClient:
    def __init__(self):
        self.channel = grpc.insecure_channel(('111.11.1.11:1111'), options=(('grpc.enable_http_proxy', 0),))
        self.stub = CalculationStub(self.channel)

    def get_result(self, json_data):
        bytes_text = bytes(json.dumps(json_data), encoding="UTF-8")
        binary_iterator = generate_text_iterator(bytes_text)
        response = self.stub.GetResult(binary_iterator)

        return response

def get_stub():
    client = GRPCClient()
    try:
        yield client.stub
    finally:
        client.channel.close()

And in the router, it was used like this.

@router.post("/", status_code=status.HTTP_201_CREATED)
def hello(stub = Depends(get_stub)):
    dic = dict(r=dict(a=1,b=3,c=5),l=dict(a=20,b=3,c=0), num=1)
    result = stub.get_result(dic)
    return

Answer 1

You're yielding client.stub, which is a CalculationStub object, but get_result is defined on the GRPCClient class. One way to solve this is to yield the whole client from your dependency:

def get_stub():
    client = GRPCClient()
    try:
        yield client
    finally:
        client.channel.close()

Another way, if you want to work directly with the CalculationStub object would be to create a custom subclass:

class MyCalculationStub(CalculationStub):
    def get_result(self, json_data):
        bytes_text = bytes(json.dumps(json_data), encoding="UTF-8")
        binary_iterator = generate_text_iterator(bytes_text)
        response = self.GetResult(binary_iterator)

        return response

and then use that in your dependency:

def get_stub():
    channel = grpc.insecure_channel(
        ('111.11.1.11:1111'),
        options=(('grpc.enable_http_proxy', 0),),
    )
    stub = MyCalculationStub(channel)
    try:
        yield stub
    finally:
        channel.close()