小编sop*_*ros的帖子

在我的功能测试中，我需要创建一个对象（与虚拟服务器相关 - 不是模拟的），该对象将由一组测试共享。服务器是使用一些 py.test 固定装置（环境、主机等）创建的，以指定应该设置的环境 - 测试在多个不同的环境上运行。我目前的方法很丑陋，就像这个抽象的例子：

# [host, env, dns are fixtures, defined in conftest.py and passed from
# commandline args - py.test recognizes them so they can be changed
# each time the test is invoked from CLI]

@pytest.mark.parametrize(parameters_from_parametrize)

def test_something(host, env, dns):
    server = Server(host,env, dns, random_name, random_passwd)
    server.create()
    server.do_stuff(parameters_from_parametrize)
    check_stuff()
    server.delete()

但这会为每组parameters_from_parametrize创建一个新服务器。有没有办法使用固定装置创建一台服务器，然后运行该服务器？

例如：

server = Server(host,env, dns, random_name, random_passwd)
server.create()

@pytest.mark.parametrize(parameters_from_parametrize)

def test_something(server):
    server.do_stuff(parameters_from_parametrize)
    check_stuff()

server.delete()

这样服务器只会被创建一次。上面的示例不起作用，因为在测试函数之前不能使用固定装置（并且测试函数外部的代码不会与它们共享范围），因此在测试函数之外添加服务器会失败。我尝试了设置和拆卸，但无法使其与固定装置一起使用。

有办法做到吗？我以为我可以使用安装/拆卸或创建测试类来完成此操作，但我失败了。

我希望这不是一个极其幼稚的问题，但很可能确实如此。

我使用concurrent.futures.ProcessPoolExecutor来查找数字范围内的数字的出现.目的是调查从并发中获得的加速性能的数量.为了测试性能,我有一个控件 - 一个执行所述任务的串行代码(如下所示).我编写了2个并发代码,一个使用concurrent.futures.ProcessPoolExecutor,另一个concurrent.futures.ProcessPoolExecutor.submit()用于执行相同的任务.它们如下所示.关于起草前者和后者的建议可分别在这里和这里看到.

发给所有三个代码的任务是在0到1E8的数字范围内找到数字5的出现次数.无论concurrent.futures.ProcessPoolExecutor.map()与.submit()被指派6名工人,并.map()有10000 CHUNKSIZE.在并发代码中,分离工作负载的方式是相同的.但是,用于在两个代码中查找出现的函数是不同的.这是因为参数传递给.submit()和.map()调用的函数的方式不同.

所有3个代码报告的发生次数相同,即56,953,279次.但是,完成任务所需的时间非常不同..map()执行速度比控制快2倍,同时控制时间.submit()是控制完成任务的两倍.

问题:

1. 我想知道.map()我的编码是否是一个神器,或者它本身就很慢？"如果是前者,我怎么能改进它.我只是惊讶它表现得比控制慢,因为没有多少激励使用它.
2. 我想知道是否还有.submit()更快的代码执行.我有一个条件是函数.map()必须返回一个包含数字5的数字/出现次数的iterable.

基准测试结果

concurrent.futures.ProcessPoolExecutor.submit()

#!/usr/bin/python3.5
# -*- coding: utf-8 -*-

import concurrent.futures as cf
from time import time
from traceback import print_exc

def _findmatch(nmin, nmax, number):
    '''Function to find the occurrence of number in range nmin to nmax and return
       the found occurrences in a list.'''
    print('\n def _findmatch', …

我使用了基于Clojure的'Anglican',我觉得这对我不好.糟糕的文件和太小的社区找不到帮助.此外,我仍然无法熟悉基于Scheme的语言.所以我想将语言改为基于Python的东西.

也许Pyro或Pymc可能就是这种情况,但我完全不知道这两者.

• 这两个框架有什么区别？
• 他们可以用于同样的问题吗？
• 有没有例子,比较闪耀的是什么？

我无法在PyYAML文档中找到如何执行此操作.我想表示我在YAML中定义的python类,如果没有在YAML中指定,则在构造函数中为参数赋予默认值.例如:

>>> class Test(yaml.YAMLObject):
...     yaml_tag = u"!Test"
...     def __init__(self, foo, bar=3):
...             self.foo = foo
...             self.bar = bar
...     def __repr__(self):
...             return "%s(foo=%r, bar=%r)" % (self.__class__.__name__, self.foo, self.bar)
... 
>>> yaml.load("""
... --- !Test
... foo: 5
... """)
Traceback (most recent call last):
  File "<stdin>", line 4, in <module>
  File "<stdin>", line 7, in __repr__
AttributeError: 'Test' object has no attribute 'bar'

我希望它会创建一个bar = 3的Test对象,但我猜它会在创建对象时绕过我的构造函数.如果我在YAML中包含bar的映射,那么一切都按预期工作:

>>> yaml.load("""
... --- !Test
... foo: 5
... bar: …

要封装我使用enum模块的状态列表：

from enum import Enum

class MyEnum(Enum):
    state1='state1'
    state2 = 'state2'

state = MyEnum.state1
MyEnum['state1'] == state  # here it works
'state1' == state  # here it does not throw but returns False (fail!)

但是，问题是我需要在脚本的许多上下文中无缝地将值用作字符串，例如：

select_query1 = select(...).where(Process.status == str(MyEnum.state1))  # works but ugly

select_query2 = select(...).where(Process.status == MyEnum.state1)  # throws exeption

如何避免调用额外的类型转换（str(state)以上）或底层值（state.value）？

所以我需要为列表中的所有项调用异步函数.这可以是一个URL列表和一个使用aiohttp的异步函数,它从每个URL获取响应.现在显然我不能做到以下几点:

['www.google.com','www.youtube.com','www.aol.com']中的url async:

我可以使用普通的for循环但是我的代码将同步操作,我失去了获得异步响应获取功能的好处和速度.

有什么方法可以转换列表,以便上述工作吗？我只需要将列表的iter()更改为aiter()方法吗？可以通过子类化列表来实现吗？也许将它封装在一个类中？

提前致谢!

(PS:我正在使用python 3.6)

除了对字符串进行"替换"链接(即text.replace(a,b).replace(c,d).replace(e,f)...)之外,是否有任何推荐的方法来进行多个字符串替换？例如,您如何实现一个快速的函数,其行为类似于Python中的PHP的htmlspecialchars？

我比较了(1)多个'替换'方法,(2)正则表达方法,和(3)马特安德森的方法.

n = 10次运行,结果如下:

在100个字符上:

TIME: 0 ms [ replace_method(str) ]
TIME: 5 ms [ regular_expression_method(str, dict) ]
TIME: 1 ms [ matts_multi_replace_method(list, str) ]

在1000个字符上:

TIME: 0 ms [ replace_method(str) ]
TIME: 3 ms [ regular_expression_method(str, dict) ]
TIME: 2 ms [ matts_multi_replace_method(list, str) ]

在10000个字符上:

TIME: 3 ms [ replace_method(str) ]
TIME: 7 ms [ regular_expression_method(str, dict) ]
TIME: 5 ms [ matts_multi_replace_method(list, str) ]

在100000个字符上:

TIME: 36 ms [ replace_method(str) ]
TIME: 46 ms [ regular_expression_method(str, …

我怀疑

airflow run dag_id task_id execution_date

会运行所有上游任务,但事实并非如此.当它看到并非所有依赖任务都运行时,它将失败.如何运行特定任务及其所有依赖项？我猜这是不可能的,因为气流设计的决定,但有没有办法解决这个问题？

best_params_train = dict(optimizer=optimizers[0], learning_rate=learning_rates[0],
                         cnn_train_type=cnn_train_types[0], 
                         cnn_arch=cnns_arch.values()[0],
                         dropout=dropouts[0])

它在 cnn_arch=cnns_arch.values()[0] 处给出错误 TypeError: 'dict_values' object is not subscriptable。我尝试转换为列表但没有成功。如何将上面的 dict(....) 转换为列表

exp_params_train = dict(optimizer=optimizers[1:], learning_rate=learning_rates[1:],
                        cnn_train_type=cnn_train_types[1:], dropout=dropouts[1:],
                        cnn_arch=cnns_arch.values())

我在python 3.3.4中使用"decode"方法遇到了一些问题.这是我的代码:

for lines in open('file','r'):
    decodedLine = lines.decode('ISO-8859-1')
    line = decodedLine.split('\t')

但是我无法解决这个问题:

AttributeError: 'str' object has no attribute 'decode'

你有什么想法？谢谢