CB-12 Use asyncio and bind application interfaces

Author: izzat5233

app/core/di/__init__.py

@@ -1,3 +1,3 @@
-from dependency_injection import get_dependency
+from .dependency_injection import get_dependency
 
 __all__ = ["get_dependency"]

app/core/di/dependency_injection.py

@@ -1,9 +1,13 @@
 from injector import Binder, Injector
 
+from app.application.generation import GenerationService, DefaultGenerationService
+from app.application.routing import RoutingService, DefaultRoutingService
+
 
 # Configure the dependency injection container
 def configure(binder: Binder) -> None:
-    # binder.bind(MyServiceInterface, to=MyService, scope=singleton)
+    binder.bind(GenerationService, to=DefaultGenerationService)
+    binder.bind(RoutingService, to=DefaultRoutingService)
     pass
 
 

app/core/interceptors/logging_interceptor.py

@@ -1,21 +1,24 @@
 import time
-from typing import Callable
-
+from typing import Callable, Any
 import grpc
-from grpc_interceptor import ServerInterceptor
 
 from app.core.logger import logger
 
 
-class LoggingInterceptor(ServerInterceptor):
-    def intercept(self, method: Callable, request: any, context: grpc.ServicerContext, method_name: str) -> any:
+class LoggingInterceptor(grpc.aio.ServerInterceptor):
+    async def intercept_service(
+            self,
+            continuation: Callable[[grpc.HandlerCallDetails], Any],
+            handler_call_details: grpc.HandlerCallDetails,
+    ) -> Any:
         start_time = time.time()
+        method_name = handler_call_details.method
         logger.info(f"Request: {method_name}")
 
         try:
-            response = method(request, context)
+            response = await continuation(handler_call_details)
             process_time = time.time() - start_time
-            logger.info(f"Response: Success (Time: {process_time:.2f}s) - {response}")
+            logger.info(f"Response: Success (Time: {process_time:.2f}s)")
             return response
         except Exception as e:
             process_time = time.time() - start_time

app/main.py

@@ -1,48 +1,75 @@
+import asyncio
 import os
+import uuid
 from concurrent import futures
 
 import grpc
 from grpc_reflection.v1alpha import reflection
 
+from app.application.generation import GenerationService
+from app.application.routing import RoutingService
+from app.core.di import get_dependency
 from app.core.interceptors import LoggingInterceptor
-from app.generated.v1.executor import service_pb2_grpc, generation_pb2, routing_pb2, service_pb2, common_pb2
-
 from app.core.logger import logger
+from app.domain.data import TextData, OptionData
+from app.domain.generation import GenerationOptions, GenerationRequest
+from app.domain.routing import RoutingRequest
+from app.generated.v1.executor import service_pb2_grpc, generation_pb2, routing_pb2, service_pb2, common_pb2
 
 
-# Service implementation
 class ExecutorService(service_pb2_grpc.ExecutorServiceServicer):
-    def Generate(self, request: generation_pb2.GenerationRequest,
-                 context: grpc.ServicerContext) -> generation_pb2.GenerationResponse:
-        text = request.input.text
-        use_memory = request.options.use_memory
-        response_schema = request.options.response_json_schema
-
-        # Your generation logic here
-        generated_text = f"Generated response for: {text}"  # Placeholder
-
-        return generation_pb2.GenerationResponse(
-            generated_output=common_pb2.TextData(text=generated_text)
+    def __init__(self):
+        self.generation_service = get_dependency(GenerationService)
+        self.routing_service = get_dependency(RoutingService)
+
+    def Generate(
+            self,
+            request: generation_pb2.GenerationRequest,
+            context: grpc.ServicerContext) -> generation_pb2.GenerationResponse:
+        request = GenerationRequest(
+            input=TextData(text=request.input.text),
+            options=GenerationOptions(
+                use_memory=request.options.use_memory,
+                response_json_schema=request.options.response_json_schema
+            ),
+            context_id=uuid.UUID(request.context_id)
         )
 
-    def Route(self, request: routing_pb2.RoutingRequest,
-              context: grpc.ServicerContext) -> routing_pb2.RoutingResponse:
-        text = request.input.text
-        options = [opt.option for opt in request.options]
-
-        # Your routing logic here
-        selected = options[0] if options else ""
-        is_fallback = False
-
-        return routing_pb2.RoutingResponse(
-            selected_option=common_pb2.OptionData(option=selected),
-            is_fallback=is_fallback
+        try:
+            result = asyncio.run(self.generation_service.generate(request))
+            logger.info(f"Generated output: {result.generated_output.text}")
+            return generation_pb2.GenerationResponse(
+                generated_output=common_pb2.TextData(text=result.generated_output.text)
+            )
+        except Exception as e:
+            return generation_pb2.GenerationResponse(
+                error=str(e)
+            )
+
+    def Route(
+            self,
+            request: routing_pb2.RoutingRequest,
+            context: grpc.ServicerContext) -> routing_pb2.RoutingResponse:
+        request = RoutingRequest(
+            input=TextData(text=request.input.text),
+            options=[OptionData(option=opt.option) for opt in request.options]
         )
 
-
-def serve():
-    server = grpc.server(
-        futures.ThreadPoolExecutor(max_workers=10),
+        try:
+            result = asyncio.run(self.routing_service.route(request))
+            logger.info(f"Selected option: {result.selected_option.option}")
+            return routing_pb2.RoutingResponse(
+                selected_option=common_pb2.OptionData(option=result.selected_option.option),
+                is_fallback=result.is_fallback
+            )
+        except Exception as e:
+            return routing_pb2.RoutingResponse(
+                error=str(e)
+            )
+
+
+async def serve():
+    server = grpc.aio.server(
         interceptors=[LoggingInterceptor()]
     )
 
@@ -63,9 +90,9 @@ def serve():
     # Start the server
     server.add_insecure_port('[::]:50051')
     logger.info("Starting Chatbot Builder Executor gRPC server on port 50051 with reflection enabled")
-    server.start()
-    server.wait_for_termination()
+    await server.start()
+    await server.wait_for_termination()
 
 
 if __name__ == '__main__':
-    serve()
+    asyncio.run(serve())