|
|
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): |
|
|
... |
|
|
|
