138 lines
No EOL
4.8 KiB
Python
138 lines
No EOL
4.8 KiB
Python
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy import create_engine
|
|
from typing import Type, Dict, Any
|
|
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
|
|
from core.models.python_models.Base import Base
|
|
from core.models.python_models.CachedSync import CachedSync
|
|
from core.models.manage.session_management import get_session
|
|
|
|
def insert_into_model(model_class: Type, all_data: dict | list, override=False) -> bool:
|
|
"""
|
|
Generic ORM insert for any model.
|
|
|
|
Args:
|
|
model_class: The ORM model class
|
|
all_data: Dictionary or list of dictionaries with data matching the model
|
|
override: If True, wipe the table before inserting
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
|
|
print(f"all_data is {all_data}")
|
|
|
|
with get_session() as session:
|
|
# Step 1: Wipe the table if override=True
|
|
if override:
|
|
session.query(model_class).delete()
|
|
session.commit()
|
|
|
|
print(f"Starting insert for {model_class.__name__}")
|
|
|
|
# Normalize to list for uniform handling
|
|
data_list = all_data if isinstance(all_data, list) else [all_data]
|
|
|
|
try:
|
|
for each_json in data_list:
|
|
instance = model_class(**each_json)
|
|
session.add(instance)
|
|
|
|
session.commit()
|
|
return True
|
|
|
|
except TypeError as e:
|
|
print(f"TypeError caught: {e}")
|
|
session.rollback()
|
|
raise ValueError(f"Invalid fields for {model_class.__name__}: {e}")
|
|
except IntegrityError as e:
|
|
print(f"IntegrityError caught: {e.orig}")
|
|
session.rollback()
|
|
raise ValueError(f"Constraint violation: {e.orig}")
|
|
except SQLAlchemyError as e:
|
|
print(f"SQLAlchemyError caught: {e}")
|
|
session.rollback()
|
|
raise
|
|
except Exception as e:
|
|
print(f"Generic exception caught: {e}")
|
|
session.rollback()
|
|
raise e
|
|
|
|
|
|
|
|
# def insert_into_model(model_class: Type, all_data: list, override=False) -> Any:
|
|
# """
|
|
# Generic ORM insert for any model. But not used for the legacy codebase yet.
|
|
|
|
# Args:
|
|
# model_class: The ORM model class
|
|
# json_data: Dictionary with data matching the model
|
|
|
|
# Returns:
|
|
# The created instance
|
|
# """
|
|
|
|
# print(f"all_data is {all_data}")
|
|
|
|
# with get_session() as session:
|
|
# # Step 1: Wipe the table if override=True
|
|
# if override:
|
|
# session.query(model_class).delete()
|
|
# session.commit()
|
|
|
|
# print("we made it past getting the session")
|
|
|
|
# print(f"Starting insert for {model_class.__name__}")
|
|
|
|
# type_of_data = type(all_data)
|
|
# print(f"all_data type is: {type_of_data}")
|
|
|
|
# if isinstance(all_data, dict):
|
|
# try:
|
|
# instance = model_class(**all_data)
|
|
# session.add(instance)
|
|
# session.commit()
|
|
# except TypeError as e:
|
|
# print(f"TypeError caught: {e}")
|
|
# session.rollback()
|
|
# raise ValueError(f"Invalid fields for {model_class.__name__}: {e}")
|
|
# except IntegrityError as e:
|
|
# print(f"IntegrityError caught: {e.orig}")
|
|
# session.rollback()
|
|
# raise ValueError(f"Constraint violation: {e.orig}")
|
|
# except SQLAlchemyError as e:
|
|
# print(f"SQLAlchemyError caught: {e}")
|
|
# session.rollback()
|
|
# raise
|
|
# except Exception as e:
|
|
# print(f"Generic exception caught: {e}")
|
|
# session.rollback()
|
|
# raise e
|
|
# return True
|
|
|
|
# elif isinstance(all_data, list):
|
|
# for each_json in all_data:
|
|
# try:
|
|
# instance = model_class(**each_json)
|
|
# session.add(instance)
|
|
# session.commit()
|
|
# except TypeError as e:
|
|
# print(f"TypeError caught: {e}")
|
|
# session.rollback()
|
|
# raise ValueError(f"Invalid fields for {model_class.__name__}: {e}")
|
|
# except IntegrityError as e:
|
|
# print(f"IntegrityError caught: {e.orig}")
|
|
# session.rollback()
|
|
# raise ValueError(f"Constraint violation: {e.orig}")
|
|
# except SQLAlchemyError as e:
|
|
# print(f"SQLAlchemyError caught: {e}")
|
|
# session.rollback()
|
|
# raise
|
|
# except Exception as e:
|
|
# print(f"Generic exception caught: {e}")
|
|
# session.rollback()
|
|
# raise e
|
|
# return True
|
|
|
|
# else:
|
|
# print("invalid data type")
|
|
# return False |