在运行时临时修改字典的Python代码实例

1 投票
5 回答
1895 浏览
提问于 2025-04-18 18:52

我在写代码的时候发现了一个模式

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..

我正在处理一些事件,这些事件会通过不同的队列传递,但每个队列的格式都有点不一样。我开始这样做:

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

但是这样看起来很丑,而且有点影响代码的运行速度。我还能做些什么呢?

5 个回答

0

我觉得下面的代码可能会解决这个问题:

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

如果你一开始只定义了那些在每种使用情况下都共同存在的键,那么你可以使用 mock 库。mock.patch.dict 这个功能可以让你在 with 语句的期间,临时向一个字典里添加键,不过你不能临时删除键。

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 是一个第三方模块,适用于 Python 2.x 版本,在 Python 3.4 之前,它被加入到了标准库中,叫做 unittest.mock

1

你可以创建一个新的字典,里面放入你想要的新字段,然后用 dict.update 方法把基础字段更新到这个新字典里。

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

你也可以用列表的方式来创建一个新的字典,里面放入字段:

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

如果你经常在大字典上这样操作,而且不想创建一个副本的话,可以使用一个叫做 上下文管理器 的东西,它可以临时修改字典,自动化你现在正在做的事情。


另外一个选择是,不用上下文管理器,而是把修改操作放在一个函数里,然后把你想要执行的操作作为函数传进去:

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 )
1

如果对字典的修改次数相对于字典本身的大小来说比较少,你可以通过创建一个上下文管理器函数来避免每次都复制字典。这样做可以确保在使用这个字典的过程中,所做的任何修改都是临时的,即使在这个过程中发生了异常,也不会影响到字典的原始内容。

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
3

假设 queue.push 只需要读取权限,你可以尝试这样做:

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)

这样做会给你一个字典,可以从两个来源获取项目,但避免了从原始数据中再创建一个实际副本的开销(不过你可能需要实现的不仅仅是 __getitem__,这取决于 push 需要什么)。

用法:

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

或者:

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

撰写回答