Redis Object Mapper (ROM) - upsert operation

rom_upsert.py

# (C) By Josiah Carlson
# https://github.com/josiahcarlson/rom/issues/116
#
def create_or_update(cls, data, **by):
    entity = cls.get_by(**by)
    if not entity:
        entity = cls(**data)
    else:
        for k,v in data.items():
            setattr(entity, k, v)
    return entity

eid = entity.get('id')
typ = "{},{}".format(_type, ", ".join(additional_types))
attrs = {'id': eid, 'type': typ}
by = {'id': eid}
mdata = create_or_update(ELock, attrs, **by)
mdata.save()

# If there is a chance of concurrent execution, and you may have two threads racing, 
# and you want to force your changes through, you can:

def force_update_data(cls, entity, attrs, _timeout=30, _limit=1000, fail_callback=None, succeed_callback=None, **by):
    _die_if_changed = attrs.pop('_die_if_changed', None)
    _die_if_mine_changed = attrs.pop('_die_if_mine_changed', None)
    geldto = lambda: {k: getattr(entity, k) for k in attrs}
    old = getold()

    ok = 0
    _timeout = time.monotonic() + max(_timeout, 0.001)
    while time.monotonic() < _timeout and _limit > 0: # or some other condition
        _limit -= 1

        try:
            mdata.save()
            ok = 1
            break

        except (rom.DataRaceError, rom.UniqueKeyViolation):
            if _die_if_changed:
                if _fail_callback:
                    _fail_callback()
                    return
                raise

            if mdata._new:
                # someone else saved this before us
                # and we were new, so get the other one
                rom.session.forget(mdata)
                mdata = create_or_update(cls, attrs, **by)
            else:
                # we were racing on this update, try again
                mdata.refresh(force=True)

            if _die_if_mine_changed and old != getold():
                if _fail_callback:
                    _fail_callback()
                    return
                raise

    else:
        if fail_callback:
            fail_callback(entity, attrs)

    if ok and succeed_callback:
        succeed_callback(entity, attrs)

force_update_data(mdata, attrs, **by)