import asyncio from fastapi_router_controller import Controller from fastapi import APIRouter, Depends, HTTPException from services.ibm_model.ibm_extract_code_block import IbmExtractCodeblock from services.ibm_model.ibm_text_generator import IbmTextGenerator from services.model_generator import ModelGenerator from utils.logger import Logger logger = Logger.get_logger(__name__) router = APIRouter(prefix='/v1') controller = Controller(router, openapi_tag={ 'name': 'Generate the Code Snippets', }) model = ModelGenerator() @controller.use() @controller.resource() class GenerateController(): def __init__( self, service: ModelGenerator = Depends()) -> None: self.model_generator = service @controller.route.get( '/generate', tags=['generate-code'], summary='Generates the code for the given input') async def generate_code(self, input: str): try: if not input: logger.error('Input is required.') raise HTTPException( status_code=500, detail='Input is required.') ibm_generate_text_visitor = IbmTextGenerator() generated_text = await self.model_generator.acceptTextGenerator(ibm_generate_text_visitor, input) ibm_extract_code_block_visitor = IbmExtractCodeblock() code_block = self.model_generator.acceptExtractCodeBlock( ibm_extract_code_block_visitor, generated_text) logger.info('Output: {}'.format(generated_text)) return {"data": code_block} except asyncio.CancelledError: logger.error( 'Canceling network request due to disconnect in client.') except Exception as error: logger.error('Error {}'.format(error))