CODEWITHSUNDEEP
python
user37203
user37203

Reputation: 706

python recipe for temporarily modifying dict on the fly

I see this pattern in some code that i'm writing

e = {...} # a dictionary
e["table"] = "users"
e["timestamp"] = time.time()
queue.push(e)
del e["table"]
del e["timestamp"]
[...]
e["table"] = "events"
queue2.push(e)
del e["table"]
# etc..

I'm demultiplexing an event over some queues but each queue has a slightly different format. I've started doing this:

queue.push( dict(e.items() + [("table":"users"), ("timestamp", time.time())]) )

but it looks ugly and it kind of slows down the code. What else can i do?

Upvotes: 1

Views: 2093

Answers (5)

Anatoliy Aksenov
Anatoliy Aksenov

Reputation: 141

I think it is might be covered with the next code:

a = {'val': 2, 'val2': -5, "name": 'Vladimir'}
b = {"asdf": 1, "b2": 2}
queue.push( dict( **a, **b) )

Upvotes: 0

martineau
martineau

Reputation: 123483

If the number of modifications to the dictionary are relatively small compared to the size of the dictionary itself, you can avoid making a copy of it each time by creating a context manager function and using it as shown. This will insure that any changes made to the dictionary are temporary, even if an exception is thrown while using it inside the block.

from contextlib import contextmanager

@contextmanager
def contextdict(adict, **kwargs):
    # modify dictionary
    changed = {}
    added = []
    for key in kwargs:
        if key in adict:
            changed[key] = adict[key]
        else:
            added.append(key)
        adict[key] = kwargs[key]
    yield adict
    # restore dictionary
    adict.update(changed)
    for key in added:
        del adict[key]

e = dict(...)  # some dictionary

with contextdict(e, table="users", timestamp=time.time()) as context:
    queue.push(context)
with contextdict(e, table="events") as context:
    queue.push(context)

# e will be unchanged at this point

Upvotes: 1

chepner
chepner

Reputation: 531345

If you initially define e with only those keys that are common to each use case, you can use the mock library. mock.patch.dict allows you to temporarily add keys to a dictionary (for the duration of the with statement), although you cannot temporarily remove keys.

e = { ... }
with mock.patch.dict(e, table="users", timestamp=time.time()):
    queue.push(e)

with mock.patch.dict(e, table="events"):
    queue2.push(e)

mock is a third-party module for Python 2.x and prior to Python 3.4, where it was added to the standard library as unittest.mock.

Upvotes: 0

loopbackbee
loopbackbee

Reputation: 23322

You can create a new dictionary with the new fields you want and use dict.update on it with the base fields

e = {...} # a dictionary
d={"table":"users", "timestamp":time.time()}
d.update(e)
queue.push(d)

You could also create a new dict with fields as a list:

e = {...} # a dictionary
queue.push( e.items() + [("table","users"), ("timestamp",time.time())] )

If you do this a lot on large dictionaries, and don't want to create a copy, you can use a Context Manager that modifies the dictionary temporarily, automating what you're doing right now.

Another option, instead of the context manager, is performing the modification in a function, passing the operations you want to do as a function:

def modify_dict_and_call( d, newfields, f):
    for k,v in newfields.items():
        d[k]=v
    f(d)
    for k in newfields:
        del d[k]

e = {...} # a dictionary
modify_dict_and_call( e, {"table":"users", "timestamp":time.time()}, queue.push )

Upvotes: 1

nmclean
nmclean

Reputation: 7734

Assuming queue.push only needs read access, you could try something like this:

class MergedDicts(dict):
    def __init__(self, *dicts, **kw):
        self.dicts = dicts + (kw,)

    def __getitem__(self, key):
        for d in self.dicts:
            if key in d: return d[key]
        raise KeyError(key)

This would give you a dictionary returning items from both sources, but avoid the overhead of building another actual copy from the originals (you may need to implement more than just __getitem__ though, depending on what push needs).

Usage:

other = {"table": "users", "timestamp": time.time()}
queue.push(MergedDicts(e, other))

or:

queue.push(MergedDicts(e, table="users", timestamp=time.time()))

Upvotes: 3

Related Questions