File size: 1,805 Bytes
589c127 8cb6721 589c127 8cb6721 589c127 8cb6721 589c127 8cb6721 589c127 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 |
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):
...
|