92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
from . import files
|
|
from jinja2 import Environment, PackageLoader
|
|
from sqlalchemy import text
|
|
from datetime import datetime, date, time
|
|
from decimal import Decimal
|
|
from .config import config as cfg
|
|
from . import files
|
|
|
|
class Modeler():
|
|
def __init__(self, src_name, filepath):
|
|
self.name = src_name
|
|
self.filepath = filepath
|
|
self.bn, self.filename, self.filetype = files.get_filename_from_path(self.filepath)
|
|
|
|
def infer_type_from_value(self, value):
|
|
if isinstance(value, int):
|
|
return 'Integer' # Maps to INTEGER in SQL
|
|
elif isinstance(value, str):
|
|
return 'String' # Maps to VARCHAR in SQL
|
|
elif isinstance(value, float):
|
|
return 'Float' # Maps to FLOAT in SQL
|
|
elif isinstance(value, Decimal):
|
|
# Infer precision and scale for decimal values
|
|
return f'Numeric(12, 3)' # Maps to NUMERIC in SQL
|
|
elif isinstance(value, datetime):
|
|
return 'DateTime' # Maps to DATETIME in SQL
|
|
elif isinstance(value, date):
|
|
return 'Date' # Maps to DATE in SQL
|
|
elif isinstance(value, time):
|
|
return 'Time' # Maps to TIME in SQL
|
|
elif isinstance(value, bool):
|
|
return 'Boolean' # Maps to BOOLEAN in SQL
|
|
elif isinstance(value, bytes):
|
|
return 'LargeBinary' # Maps to BLOB or BYTEA in SQL
|
|
else:
|
|
return 'String' # Default fall-back type
|
|
|
|
def render_model(self, columns):
|
|
version = datetime.now().strftime("%Y%m%d%H%M%S")
|
|
|
|
try:
|
|
env = Environment(loader=PackageLoader('dft', 'templates'))
|
|
template = env.get_template('model_declaration.jinja2')
|
|
|
|
rendered_content = template.render(
|
|
from_file = f"{(version)}_{str(self.filepath)}",
|
|
migration_path=cfg["migration-path"],
|
|
model_name=self.filename,
|
|
revision_id=version,
|
|
create_date=datetime.now().isoformat(),
|
|
columns=columns
|
|
)
|
|
except Exception as e:
|
|
return e
|
|
else:
|
|
print(f"Model {self.filename} content is rendered")
|
|
return rendered_content
|
|
|
|
def write_model_to_file(self, content):
|
|
to_file = cfg["migration-path"] + f"/models/{self.filename}.py"
|
|
try:
|
|
# Write the rendered content to a file
|
|
with open(to_file, 'w') as f:
|
|
f.write(content)
|
|
except Exception as e:
|
|
return e
|
|
else:
|
|
print(f"Model {self.filename} has been written to file {to_file} with success !")
|
|
return True
|
|
|
|
|
|
class SQLModeler(Modeler):
|
|
|
|
def __init__(self, src_name, filepath):
|
|
super().__init__(src_name, filepath)
|
|
|
|
def infer_model_columns(self, result):
|
|
first_row = result.fetchone()
|
|
columns = []
|
|
for column, value in zip(result.keys(), first_row):
|
|
sql_type = self.infer_type_from_value(value)
|
|
column = f'{column} = Column("{column}", {sql_type})'
|
|
columns.append(column)
|
|
return columns
|
|
|
|
def generate_model(self, query):
|
|
with self._create_engine().connect() as cnt:
|
|
rst = cnt.execute(text(query))
|
|
columns = self.infer_model_columns(rst)
|
|
content = self.render_model(columns)
|
|
self.write_model_to_file(content)
|