66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
from core.errors.logger import logger
|
||
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.manage.session_management import get_session
|
||
from core.models.orm_models.Base import Base
|
||
|
||
# all models it knows how to do:
|
||
from core.models.orm_models.CachedSync import CachedSync
|
||
from core.models.orm_models.Location import Location
|
||
from core.models.orm_models.Operator import Operator
|
||
from core.models.orm_models.EncryptedProxy import EncryptedProxy
|
||
|
||
def insert_into_model(model_class: Type, all_data: dict | list, override=False) -> bool:
|
||
"""
|
||
Generic ORM insert for any model. Keeping it generic for reuse.
|
||
|
||
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
|
||
"""
|
||
|
||
logger.info(f"All the public data preparing to be inserted is {all_data}")
|
||
|
||
with get_session() as session:
|
||
# Step 1: Wipe the table if override=True
|
||
if override:
|
||
logger.info(f"First, WIPING the pre-existing data for {model_class.__name__}. Are you sure you intended to completely delete the old data?")
|
||
session.query(model_class).delete()
|
||
session.commit()
|
||
|
||
logger.info(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()
|
||
logger.info(f"Completed! SQL Insertion Successful following the {model_class.__name__} Model’s Type Rules")
|
||
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
|