57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
#helper/db_adapter
|
|
|
|
from typing import List, Dict
|
|
from database import sessionLocal
|
|
from app.modules.edos.models import EDOS
|
|
|
|
class DBAdapter:
|
|
""" Adapter: insert cleen data on a db"""
|
|
|
|
def __init__(self):
|
|
self.db = sessionLocal()
|
|
|
|
def insert_filtered_data(self, mapped_rows: List[Dict]) -> Dict:
|
|
""" Insert only data filtered and mapped"""
|
|
print(f" [DBAdapter] Recibidas {len(mapped_rows)} filas")
|
|
|
|
if mapped_rows:
|
|
print(f" [DBAdapter] Primera fila: {mapped_rows[0]}")
|
|
|
|
|
|
|
|
results = {
|
|
'total_processed': len(mapped_rows),
|
|
'inserted': 0,
|
|
'errors': [],
|
|
'duplicates_skiped': 0
|
|
}
|
|
for row in mapped_rows:
|
|
try:
|
|
#verified dupled data
|
|
|
|
unique_field_value = row.get('numero')
|
|
|
|
existing = self.db.query(EDOS).filter_by(
|
|
numero=unique_field_value
|
|
).first()
|
|
|
|
if existing:
|
|
results['duplicates_skiped'] += 1
|
|
continue
|
|
|
|
instance = EDOS(**row)
|
|
|
|
self.db.add(instance)
|
|
print(f" [DBAdapter] Added instance for {row.get('numero')}")
|
|
self.db.commit()
|
|
print(f" [DBAdapter] Committed successfully")
|
|
results['inserted'] += 1
|
|
except Exception as e:
|
|
self.db.rollback()
|
|
results['errors'].append({
|
|
'row': row,
|
|
'error': str(e)
|
|
})
|
|
self.db.close()
|
|
return results
|
|
|