在运行时临时修改字典的Python代码实例
我在写代码的时候发现了一个模式
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()))