How do you manage a database connection in grpc?

Question

I'm currently following some quick start tutorials in grpc adding a bit with database connections, and obviously creating a database connection every request is not optimal

def connection():
    conn = psycopg2.connect(
        user="postgres", password="some_password", database="some_db")
    return conn


class LeagueGameManager(start_pb2_grpc.GameManagerServicer):
    async def CreateLGGame(self, request, context):
        try:
            conn = connection()
            cursor = conn.cursor()

            cursor.execute("some sql statement")

            conn.commit()
            cursor.close()
            conn.close()
        except OperationalError as e:
            context.set_details(e)
            context.set_code(grpc.StatusCode.INTERNAL)
            cursor.close()
            conn.close()
            return

        return start_pb2.GameReply(json_response=json.dumps(new_row[0]))


async def serve():
    server = grpc.aio.server()
    start_pb2_grpc.add_GameManagerServicer_to_server(
        LeagueGameManager(), server)
    listen_addr = '[::]:50051'
    server.add_insecure_port(listen_addr)
    logging.info("Starting server on %s", listen_addr)
    await server.start()
    await server.wait_for_termination()

what is the optimal way of managing a database connection above?

Answer 1

I know I'm a bit late, but due to the fact that the gRPC in the python is not well documented, I want to help future visitors. I've been struggling with gRPC recently. For me, the easiest way to have constant connection to the database was to connect to it in the constructor of the Service class. In my case I make connection to MongoDb, It should look like this:

class MyServiceClass(MyServiceClass_pb2_grpc.MyServiceClassServicer):
    def __init__(self) -> None:
        super().__init__()
        self.client = pymongo.MongoClient(CONNSTRING) # Connection to Mongo database
        self.client.server_info()
    async def get_info(self, request, context):
        response_info = MyScript(request.code).get_something()  # Call 
        return MyServiceClasse_pb2.InfoResponse(info= response_info)
async def serve():
    server = grpc.aio.server()
    start_pb2_grpc.add_MyServiceClass_to_server(
        MyServiceClass(), server)
    listen_addr = '[::]:50051'
    server.add_insecure_port(listen_addr)
    logging.info("Starting server on %s", listen_addr)
    await server.start()
    await server.wait_for_termination()