如何使 2.7 python 上下文管理器线程安全

2024-01-02

我有一个在 Django 服务上运行的大型 python 应用程序。我需要关闭某些操作的权限测试,因此我创建了这个上下文管理器:

class OverrideTests(object):

    def __init__(self):
        self.override = 0

    def __enter__(self):
        self.override += 1

    # noinspection PyUnusedLocal
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.override -= 1
        assert not self.override < 0

    @property
    def overriding(self):
        return self.override > 0

override_tests = OverrideTests()

然后,应用程序的各个部分可以使用上下文管理器覆盖测试:

with override_tests:
    do stuff
    ...

在 do stuff 中,上述上下文管理器可以在不同的函数中多次使用。计数器的使用可以控制这种情况,并且看起来工作正常......直到线程介入。

一旦涉及到线程,全局上下文管理器就会被重新使用,因此,测试可能会被错误地覆盖。

这是一个简单的测试用例 - 如果thread.start_new_thread(do_id, ())线被替换为简单的do_it但严重失败,如下所示:

def stat(k, expected):
    x = '.' if override_tests.overriding == expected else '*'
    sys.stdout.write('{0}{1}'.format(k, x))


def do_it_inner():
    with override_tests:
        stat(2, True)
    stat(3, True)  # outer with context makes this true


def do_it():
    with override_tests:
        stat(1, True)
        do_it_inner()
    stat(4, False)


def do_it_lots(ntimes=10):
    for i in range(ntimes):
        thread.start_new_thread(do_it, ())

如何使这个上下文管理器线程安全,以便在每个 Python 线程中,即使它是可重入的,也能一致地使用它?


这是一种方法seems工作:使您的 OverrideTests 类成为以下类的子类threading.local。为了安全起见,你应该调用超类__init__在你的__init__(尽管即使你不这样做它似乎也能工作):

class OverrideTests(threading.local):

    def __init__(self):
        super(OverrideTests, self).__init__()
        self.override = 0

    # rest of class same as before

override_tests = OverrideTests()

Then:

>>> do_it_lots()
1.1.1.2.2.1.1.1.1.1.1.3.3.2.2.2.2.2.2.4.4.3.1.3.3.3.3.4.3.2.4.4.2.4.3.4.4.4.3.4.

但是,我不会把钱花在这种不会在某种极端情况下失败的情况上,特别是如果您的实际应用程序比您在此处显示的示例更复杂的话。最终,您确实应该重新考虑您的设计。在您的问题中,您关注的是如何“使上下文管理器线程安全”。但真正的问题不仅在于你的上下文管理器,还在于你的function (stat在你的例子中)。stat依赖于全局状态(全局override_tests),这在线程环境中本质上是脆弱的。

本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

如何使 2.7 python 上下文管理器线程安全 的相关文章

  • 从字符串到类型的词法转换

    最近 我尝试用Python存储和读取文件中的信息 遇到了一个小问题 我想从文本文件中读取类型信息 从 string 到 int 或 float 的类型转换非常有效 但从 string 到 type 的类型转换似乎是另一个问题 当然 我尝试了
  • 可以在 TensorFlow 中使用排名相关作为成本函数吗?

    我正在处理偶尔充满异常值的极其嘈杂的数据 因此我主要依靠相关性来衡量我的神经网络的准确性 是否可以明确使用诸如等级相关性 斯皮尔曼相关系数 之类的东西作为我的成本函数 到目前为止 我主要依赖 MSE 作为相关性的代理 我现在面临三个主要障碍
  • Paramiko - 使用私钥连接 - 不是有效的 OPENSSH 私钥/公钥文件

    我正在尝试找到解决方案 但无法理解我做错了什么 在我的 Linux 服务器上 我运行了以下命令 ssh keygen t rsa 这产生了一个id rsa and id rsa pub file 然后我将它们复制到本地并尝试运行以下代码 s
  • PIL Image.size 返回相反的宽度/高度

    使用PIL确定图像的宽度和高度 在特定图像上 幸运的是只有这一个 但这很麻烦 从 image size 返回的宽度 高度是相反的 图片 http storage googleapis com cookila 533ebf752b9d1f7c
  • 带图像的简单 GUI [关闭]

    很难说出这里问的是什么 这个问题是含糊的 模糊的 不完整的 过于宽泛的或修辞性的 无法以目前的形式得到合理的回答 如需帮助澄清此问题以便重新打开 访问帮助中心 help reopen questions 我试图在简单的 GUI 上显示一些卡
  • 如何从 Dockerfile 安装 Python 3.7 和 Pip

    我正在尝试构建基于 Ubuntu 18 04 的自定义 Docker 映像 Ubuntu 预装了 Python 3 6 但我想 1 安装 Python 3 7 2 将其设置为默认 Python 版本 这样就可以使用python代替pytho
  • 定义函数后对其进行修饰?

    I think答案是否定的 但我似乎找不到明确的说法 我有以下情况 def decorated function function functools wraps function def my function print Hello s
  • 如何在 Python 中将彩色输出打印到终端?

    是否有与 Perl 等效的 Python 语言 print color red print
  • 如何在python中检索aws批处理参数值?

    流程 Dynamo DB gt Lambda gt 批处理 如果将角色 arn 插入动态数据库 它是从 lambda 事件中检索的 然后使用submit job角色 arn 的 API 被传递为 parameters role arn ar
  • spacy 如何使用词嵌入进行命名实体识别 (NER)?

    我正在尝试使用以下方法训练 NER 模型spaCy识别位置 人 名和组织 我试图理解如何spaCy识别文本中的实体 但我无法找到答案 从这个问题 https github com explosion spaCy issues 491在 Gi
  • 具有多个元素的数组的真值是二义性错误吗? Python

    from numpy import from pylab import from math import def TentMap a x if x gt 0 and x lt 0 5 return 2 a x elif x gt 0 5 a
  • Python:计算数据帧列中所有行中特定字符的实例数

    我有一个包含列 toaddress ccaddress body 的数据框 df 我想迭代数据帧的索引 以获取 toaddress 和 ccaddress 字段中电子邮件地址的最小 最大和平均数量 这是通过计算这两列中每个字段中的 和 的实
  • WPF 从主线程以外的其他线程截屏

    我有一个线程用于侦听 WPF 应用程序的命令 如果 WPF 应用程序收到截取屏幕截图的命令 则任务将移交给 screenshotService 我在互联网上的某个地方找到了一些代码来截取屏幕截图 似乎可以工作 但我还没有想清楚 我无法从另一
  • 将输入发送到 python 子进程而不等待结果

    我正在尝试为一段代码编写一些基本测试 该代码通常通过 stdin 无休止地接受输入 直到给出特定的退出命令 我想检查程序是否在给出一些输入字符串时崩溃 经过一段时间来考虑处理 但似乎无法弄清楚如何发送数据而不是陷入等待我不知道的输出关心 我
  • 线程、进程和 Application.Exit()

    我的应用程序由主消息循环 GUI 和线程 Task Factory 组成 在线程中我调用一些第三方应用程序var p new Process 但是当我调用Application Exit 在消息循环中 我可以看到在线程中启动的进程仍在内存中
  • 如何在与应用程序初始化文件不同的文件中迭代 api 路由

    我有一个 apiroutes py 文件 其中定义了许多路由 例如 api route api read methods GET api route api write methods POST 其中 api 是导入 from import
  • 如何通过字符串匹配加速 pandas 行过滤?

    我经常需要过滤 pandas 数据框df by df df col name string value 并且我想加快行选择操作 有没有快速的方法可以做到这一点 例如 In 1 df mul df 3000 2000 3 reset inde
  • Flask WTForms 使用变量自动填充 StringField

    我有一个表格 我想用上一页收到的信息自动填充一些字段 但如果他们想调整它 它需要是可更改的 我正在为我的 SelectField 使用动态创建的列表 但添加 StringField 并不成功 请参阅下面的我的代码 forms py clas
  • Python组合目录中的所有csv文件并按日期时间排序

    我有 2 年的每日数据分成每月文件 我想将所有这些数据合并到一个按日期和时间排序的文件中 我正在使用的代码组合了所有文件 但不按顺序 我正在使用的代码 import pandas as pd import glob os import cs
  • 如何禁止 celery 中的 pickle 序列化

    Celery 默认使用 pickle 作为任务的序列化方法 如中所述FAQ http ask github com celery faq html isn t using pickle a security concern 这代表一个安全漏

随机推荐