123 lines
5.2 KiB
Python
123 lines
5.2 KiB
Python
from flask import Blueprint, request, jsonify
|
|
from flask_jwt_extended import jwt_required
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy import create_engine, Column, Integer, String, Boolean, ForeignKey, DateTime, Date, ForeignKeyConstraint
|
|
from sqlalchemy import Table, MetaData, ForeignKey
|
|
from flasgger.utils import swag_from
|
|
from models.user import User
|
|
from models.dinamic_table import DynamicTable
|
|
from models.dinamic_column import DynamicColumn
|
|
from datetime import datetime
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy import inspect
|
|
from extensions import db
|
|
|
|
# Criando a base
|
|
Base = declarative_base()
|
|
|
|
# Função para criar tabelas dinamicamente com base no JSON
|
|
# Função para mapear uma tabela existente
|
|
def map_existing_table(table_name, db_engine, existing_tables):
|
|
if table_name not in existing_tables:
|
|
return None
|
|
# Usando Table do SQLAlchemy para mapear a tabela existente
|
|
metadata = MetaData(bind=db_engine)
|
|
return Table(table_name, metadata, autoload_with=db_engine)
|
|
|
|
def create_dynamic_table(json_data, db_engine):
|
|
try:
|
|
# Nome da tabela
|
|
table_name = json_data['table_name']
|
|
|
|
# Verificar se a tabela já existe no banco de dados
|
|
inspector = inspect(db_engine)
|
|
existing_tables = inspector.get_table_names()
|
|
if table_name in existing_tables:
|
|
raise ValueError(f"A tabela '{table_name}' já existe no banco de dados.")
|
|
|
|
# Mapear tabelas existentes que são referenciadas como chave estrangeira
|
|
def map_existing_table(table_name, db_engine, existing_tables):
|
|
if table_name not in existing_tables:
|
|
return None
|
|
metadata = MetaData(bind=db_engine)
|
|
return Table(table_name, metadata, autoload_with=db_engine)
|
|
|
|
# Criar a classe dinamicamente com base nas colunas fornecidas
|
|
table_class = type(table_name, (Base,), {
|
|
'__tablename__': table_name,
|
|
'id': Column(Integer, primary_key=True) # Adicionar a chave primária
|
|
})
|
|
|
|
# Processar as colunas
|
|
foreign_keys = []
|
|
for column in json_data['columns']:
|
|
column_name = column['column_name']
|
|
column_type = column['column_type']
|
|
is_nullable = column.get('is_nullable', True)
|
|
is_foreign_key = column.get('is_foreign_key', False)
|
|
related_table = column.get('related_table')
|
|
|
|
# Mapeamento de tipos
|
|
if column_type == "Integer":
|
|
sqlalchemy_type = Integer
|
|
elif column_type == "String":
|
|
sqlalchemy_type = String(255)
|
|
elif column_type == "Date":
|
|
sqlalchemy_type = Date
|
|
elif column_type == "Boolean":
|
|
sqlalchemy_type = Boolean
|
|
else:
|
|
raise ValueError(f"Tipo de coluna {column_type} não suportado")
|
|
|
|
# Verificar se a coluna é uma chave estrangeira
|
|
if is_foreign_key:
|
|
# Caso a tabela seja uma tabela existente com classe
|
|
referenced_table_class = map_existing_table(related_table, db_engine, existing_tables)
|
|
if referenced_table_class:
|
|
# Se a tabela referenciada já existir como uma classe
|
|
column_object = Column(
|
|
column_name,
|
|
sqlalchemy_type,
|
|
ForeignKey(f'{related_table}.id'), # Definindo chave estrangeira com o nome da tabela
|
|
nullable=is_nullable
|
|
)
|
|
foreign_keys.append(ForeignKeyConstraint(
|
|
[column_name], [f'{related_table}.id']
|
|
))
|
|
else:
|
|
raise ValueError(f"A tabela referenciada '{related_table}' não existe no banco de dados.")
|
|
else:
|
|
column_object = Column(column_name, sqlalchemy_type, nullable=is_nullable)
|
|
|
|
# Adicionar a coluna à classe
|
|
setattr(table_class, column_name, column_object)
|
|
|
|
# Criar a tabela no banco de dados (passando o engine para o método create_all)
|
|
Base.metadata.create_all(bind=db_engine) # Passando o engine diretamente
|
|
|
|
# Criar as chaves estrangeiras, após a criação das tabelas
|
|
for fk in foreign_keys:
|
|
# Após criar a tabela, aplicamos as chaves estrangeiras
|
|
Base.metadata.create_all(bind=db_engine)
|
|
|
|
return table_class
|
|
|
|
except SQLAlchemyError as e:
|
|
print(f"Erro ao criar a tabela dinâmica: {str(e)}")
|
|
return None
|
|
dinamic_table_bp = Blueprint('dinamic_table', __name__, url_prefix='/dinamic_table')
|
|
|
|
# Endpoint para criar a tabela dinamicamente via requisição REST
|
|
@dinamic_table_bp.route('/create_table', methods=['POST'])
|
|
@swag_from('../docs/dinamic_table/create.yml')
|
|
def create_table():
|
|
# Pega o JSON da requisição
|
|
json_data = request.get_json()
|
|
|
|
# Cria a tabela dinamicamente
|
|
try:
|
|
table_class = create_dynamic_table(json_data, db.engine)
|
|
return jsonify({"message": f"Tabela {json_data['table_name']} criada com sucesso!"}), 201
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 400
|