client.py

from abc import ABC, abstractmethod
from typing import List
from loguru import logger
import openai


class SQLService(ABC):

    @abstractmethod
    def translate(self, source_db: str, target_db: str, sql: str) -> List:
        ...


class OpenAIService(SQLService):

    def __init__(self, api_key) -> None:
        super().__init__()
        openai.api_key = api_key


    def translate(self, source_db: str, target_db: str, sql: str) -> List:
        results = []
        try:
            response = openai.Completion.create(
                model="text-davinci-003",
                prompt=f"##### Translate this function  from Oracle into Postgresql\n"
                       f"### {source_db}"
                       f""
                       f"   {sql}"
                       f""
                       f"### {target_db}",
                temperature=0,
                max_tokens=2048,
                top_p=1,
                frequency_penalty=0,
                presence_penalty=0,
                stop=["###"]
            )
            for choice in response.choices:
                logger.info(f"transform {source_db} to {target_db}, SQL:")
                logger.debug(choice.text)
                results.append(choice.text)
        except Exception as ex:
            logger.error(f"transform from {source_db} to {target_db}, failed \n {sql}")
            logger.exception(ex)
        return results


class GoogleT5Service(SQLService):

    def translate(self, source_db: str, target_db: str, sql: str) -> List:
        return ["In Developing now"]

    def __init__(self):
        ...


class FacebookLLAMAService(SQLService):

    def translate(self, source_db: str, target_db: str, sql: str) -> List:
        return ["In facebook waiting list"]

    def __init__(self):
        ...