# (c) cavaliba.com - eav from celery import shared_task from app_data.models import DataEAV, DataInstance from app_data.schema import Schema from app_home.log import INFO, TRACE, log # -------------------------------------------------------- # EAV batch state - V4.0 # -------------------------------------------------------- class EavBatch: mode = "bulk" # "single" | "bulk" | "multibulk" batch_size = 500 _batch = [] @classmethod def init(cls): cls._batch = [] @classmethod def flush(cls): if cls._batch: DataEAV.objects.bulk_create(cls._batch, ignore_conflicts=True) cls._batch = [] @classmethod def save(cls, instance): if cls.mode == "bulk": cls.save_bulk(instance) elif cls.mode == "multibulk": cls.save_bulk_multi(instance) else: cls.save_single(instance) @classmethod def save_single(cls, instance): DataEAV.objects.filter( classname=instance.classname, keyname=instance.keyname, ).delete() if not instance.iobj: return eavobj = DataEAV( iid=instance.id, classname=instance.classname, keyname=instance.keyname, displayname=instance.displayname, fieldname="keyname", format="string", value=instance.keyname, last_update=instance.last_update, ) try: eavobj.save() except Exception: pass for fieldname, field in instance.fields.items(): if field.is_injected or field.dataformat == "external": continue values = field.get_eav_list() fmt = field.get_eav_format() for v in values: if len(v) > 1000: continue if len(v) == 0: continue eavobj = DataEAV( iid=instance.id, classname=instance.classname, keyname=instance.keyname, displayname=instance.displayname, fieldname=fieldname, format=fmt, value=v, last_update=instance.last_update, ) try: eavobj.save() except Exception: pass @classmethod def save_bulk(cls, instance): DataEAV.objects.filter( classname=instance.classname, keyname=instance.keyname, ).delete() if not instance.iobj: return common = dict( iid=instance.id, classname=instance.classname, keyname=instance.keyname, displayname=instance.displayname, last_update=instance.last_update, ) objs = [DataEAV(fieldname="keyname", format="string", value=instance.keyname, **common)] for fieldname, field in instance.fields.items(): if field.is_injected or field.dataformat == "external": continue fmt = field.get_eav_format() for v in field.get_eav_list(): if not v or len(v) > 1000: continue objs.append(DataEAV(fieldname=fieldname, format=fmt, value=v, **common)) DataEAV.objects.bulk_create(objs, ignore_conflicts=True) @classmethod def save_bulk_multi(cls, instance): DataEAV.objects.filter( classname=instance.classname, keyname=instance.keyname, ).delete() if not instance.iobj: return common = dict( iid=instance.id, classname=instance.classname, keyname=instance.keyname, displayname=instance.displayname, last_update=instance.last_update, ) cls._batch.append( DataEAV(fieldname="keyname", format="string", value=instance.keyname, **common) ) for fieldname, field in instance.fields.items(): if field.is_injected or field.dataformat == "external": continue fmt = field.get_eav_format() for v in field.get_eav_list(): if not v or len(v) > 1000: continue cls._batch.append(DataEAV(fieldname=fieldname, format=fmt, value=v, **common)) if len(cls._batch) >= cls.batch_size: cls.flush() # ----------------------------------------------------------- def batch_qs(qs, batch_size=200): """ Returns a (start, end, total, queryset) tuple for each batch in the given queryset. """ total = qs.count() for start in range(0, total, batch_size): end = min(start + batch_size, total) yield (start, end, total, qs[start:end]) def eav_from_field(format=None, value=None): """ Find EAV entry by format and value E.g. : format = 'file', value = 'f390fcc8-818c-45c5-9da7-4d714434d75f' => EAV object having this file as an attachment (field value) """ if not format: return if not value: return eav = DataEAV.objects.filter(format=format, value=value).first() return eav # ---- @shared_task(ignore_result=True) def task_eav_purge(verbose=False, dryrun=True, progress=False): """aysnc to celery eav_purge""" eav_purge(verbose=verbose, dryrun=dryrun, progress=progress) def eav_purge(verbose=False, dryrun=True, progress=False): """Loop over EAV object, remove if orphan <=> not in Instance""" count_total = 0 count_delete_total = 0 classes = DataEAV.objects.values("classname").distinct() for cobj in classes: classname = cobj["classname"] instances = DataEAV.objects.filter(classname=classname).values("keyname").distinct() for start, end, total, qs in batch_qs(instances): count = 0 count_delete = 0 if progress: print( f"eav_purge - {classname} - BATCH - processing {start + 1} - {end} of {total}" ) for iobj in qs: count_total += 1 count += 1 keyname = iobj["keyname"] if verbose: log(TRACE, app="data", view="eav_purge", data=f"instance: {keyname}") found = DataInstance.objects.filter(classname=classname, keyname=keyname).count() if found == 0: if not dryrun: DataEAV.objects.filter(classname=classname, keyname=keyname).delete() count_delete_total += 1 count_delete += 1 if verbose: print(f"eav_purge - {classname} - {keyname} - orphan - deleted from EAV") log( INFO, app="data", view="eav_purge", data=f"classname: {classname} - deleted {count_delete}/{count}", ) log( INFO, app="data", view="eav_purge", data=f"TOTAL - deleted {count_delete_total}/{count_total}", ) print("COUNT FOUND: ", count_total) print("COUNT DELETED: ", count_delete_total) return count_delete_total, count_total @shared_task(ignore_result=True) def task_eav_refresh(verbose=False, dryrun=True, force=False, progress=False): """aysnc to celery eav_refresh""" eav_refresh(verbose=verbose, dryrun=dryrun, force=force, progress=progress) def eav_refresh(verbose=False, dryrun=True, force=False, progress=False): from app_data.data import Instance count = 0 count_create = 0 count_update = 0 count_action = 0 # loop over SCHEMA/INSTANCE >> refresh EAV for classname in Schema.listall_names(): if verbose: print("CLASSNAME: ", classname) instances = ( DataInstance.objects.filter(classname=classname) .values("keyname", "last_update") .order_by("id") ) for start, end, total, qs in batch_qs(instances): if progress: print( f"eav_refresh - {classname} - BATCH - processing {start + 1} - {end} of {total}" ) for item in qs: count += 1 action = False keyname = item["keyname"] last_update1 = item["last_update"] eav = ( DataEAV.objects.filter(classname=classname, keyname=item["keyname"]) .values("last_update") .first() ) if eav: last_update2 = eav["last_update"] # UPDATE if last_update1 != last_update2: count_update += 1 action = True else: # nothing to do pass else: # CREATE last_update2 = "NA" count_create += 1 action = True if verbose: print(f" {classname}::{keyname}") print(f" {last_update1} (DB)") print(f" {last_update2} (EAV)") print(f" create/update: {action}") if not dryrun: if force or action: instance = Instance.from_keyname(classname=classname, keyname=keyname) if instance: # update > rewrite EAV # instance.save() EavBatch.save(instance) count_action += 1 print("COUNT FOUND: ", count) print("COUNT CREATE NEEDED: ", count_create) print("COUNT UPDATE NEEDED: ", count_update) print("COUNT ACTION DONE: ", count_action) log(INFO, app="data", view="eav_refresh", data=f"TOTAL - created {count_create}/{count}") log(INFO, app="data", view="eav_refresh", data=f"TOTAL - updated {count_update}/{count}") return count_create, count_update, count