小编Ran*_*dra的帖子

我已经为我们的网络应用安排了一些芹菜节拍的重复任务

应用程序本身是使用金字塔Web框架构建的.使用zopetransaction扩展来管理会话

在芹菜中,我使用该应用程序作为库.我正在重新定义具有函数的模型中的会话.

它运作良好,但偶尔,它会提升 InvalidRequestError: This session is in 'prepared' state; no further SQL can be emitted within this transaction

我不确定有什么问题,为什么会发出这些警告.

示例代码:

在 tasks.py

def initialize_async_session():
    import sqlalchemy
    from webapp.models import Base, set_dbsession, engine

    Session = sqlalchemy.orm.scoped_session(
                sqlalchemy.orm.sessionmaker(autocommit=True, autoflush=True)
                            )
    Session.configure(bind=engine)
    session = Session()

    set_dbsession(session)
    Base.metadata.bind = engine
    return session


@celery.task
def rerun_scheduler():
    log.info("Starting pipeline scheduler")
    session = initialize_async_session()
    webapp.sheduledtask.service.check_for_updates(session)
    log.info("Ending pipeline scheduler")

在models.pywebapp中

DBSession = scoped_session(sessionmaker(bind=engine, expire_on_commit=False,
                    extension=ZopeTransactionExtension()))

def set_dbsession(db_session=None):
    """
    This function sets the db session …

我正在处理的应用程序非常异步.Web应用程序根据用户操作通过芹菜运行许多任务.芹菜任务本身能够启动进一步的任务.

下面显示的代码经常出现在我们的代码库中.

def do_sth():
    logic();
    if condition:
         function1.apply_async(*args)
    else:
         function2.apply_asynch(*args)

现在我们要开始对我们编写的任何新代码进行单元测试,但我们不确定如何执行此操作.我们想在我们的断言pytest单元测试是我们希望看到的,如果功能1实际上得到调用.我们不希望自己运行,function1因为我们将进行单元测试function1.

我不希望将celery作为进程运行,也不想在单元测试会话期间运行任何AMQP代理.

这可以实现吗？

编辑

有人指出,这是一个副本如何单元测试Celery任务？

它不是.想一想.我要问的是如何测试函数是否通过apply_async 调用了function1.那个问题是我如何测试function1本身.有一个主要的区别.在构建这个问题之前,我确实遇到了这个问题.

我的代码使用pytest相当多.示例代码结构如下所示.整个代码库是python-2.7

core/__init__.py
core/utils.py

#feature

core/feature/__init__.py
core/feature/service.py

#tests
core/feature/tests/__init__.py
core/feature/tests/test1.py
core/feature/tests/test2.py
core/feature/tests/test3.py
core/feature/tests/test4.py
core/feature/tests/test10.py

本service.py看起来是这样的:

from modules import stuff
from core.utils import Utility


class FeatureManager:
    # lots of other methods
    def execute(self, *args, **kwargs):
        self._execute_step1(*args, **kwargs)
        # some more code
        self._execute_step2(*args, **kwargs)
        utility = Utility()
        utility.doThings(args[0], kwargs['variable'])

所有测试feature/tests/*最终都使用了core.feature.service.FeatureManager.execute功能.但是utility.doThings(),在运行测试时,我没有必要运行.我需要它在生产应用程序运行时发生,但我不希望它在测试运行时发生.

我可以做我这样的事 core/feature/tests/test1.py

from mock import patch

class Test1:
   def test_1():
       with patch('core.feature.service.Utility') as MockedUtils:
           exectute_test_case_1()

这会奏效.但是我Utility刚刚添加到代码库中,我有超过300个测试用例.我不想进入每个测试用例并写下这个with语句.

我可以写一个conftest.py设置一个os级环境变量,根据该环境变量core.feature.service.FeatureManager.execute …

我正在使用 zeep 调用 SOAP Web 服务。即使 WSDL 中存在该方法，它也会引发错误

client = Client(self.const.soap_url)
client.service.getPlansDetails(id)

我收到这个错误

AttributeError: Service has no operation 'getPlansDetails'

以下是来自python -m zeep <wsd_url>

Prefixes:
     xsd: http://www.w3.org/2001/XMLSchema
     ns0: http://tempuri.org/Imports
     ns1: http://tempuri.org
     ns2: http://schemas.datacontract.org/2004/07/Dynamics.Ax.Application
     ns3: http://schemas.datacontract.org/2004/07/Microsoft.Dynamics.Ax.Xpp
     ns4: http://schemas.microsoft.com/dynamics/2010/01/datacontracts
     ns5: http://schemas.microsoft.com/2003/10/Serialization/Arrays
     ns6: http://schemas.microsoft.com/dynamics/2008/01/documents/Fault
     ns7: http://schemas.datacontract.org/2004/07/Microsoft.Dynamics.AX.Framework.Services
     ns8: http://schemas.microsoft.com/2003/10/Serialization/

Global elements:
     ns2:LFCPaymentPlanDetailsContract(ns2:LFCPaymentPlanDetailsContract)
     ns7:ArrayOfInfologMessage(ns7:ArrayOfInfologMessage)
     ns7:InfologMessage(ns7:InfologMessage)
     ns7:InfologMessageType(ns7:InfologMessageType)
     ns3:XppObjectBase(ns3:XppObjectBase)
     ns5:ArrayOfKeyValueOfstringstring(ns5:ArrayOfKeyValueOfstringstring)
     ns8:QName(xsd:QName)
     ns8:anyType(None)
     ns8:anyURI(xsd:anyURI)
     ns8:base64Binary(xsd:base64Binary)
     ns8:boolean(xsd:boolean)
     ns8:byte(xsd:byte)
     ns8:char(ns8:char)
     ns8:dateTime(xsd:dateTime)
     ns8:decimal(xsd:decimal)
     ns8:double(xsd:double)
     ns8:duration(ns8:duration)
     ns8:float(xsd:float)
     ns8:guid(ns8:guid)
     ns8:int(xsd:int)
     ns8:long(xsd:long)
     ns8:short(xsd:short)
     ns8:string(xsd:string)
     ns8:unsignedByte(xsd:unsignedByte)
     ns8:unsignedInt(xsd:unsignedInt)
     ns8:unsignedLong(xsd:unsignedLong)
     ns8:unsignedShort(xsd:unsignedShort)
     ns6:AifFault(ns6:AifFault)
     ns6:ArrayOfFaultMessage(ns6:ArrayOfFaultMessage)
     ns6:ArrayOfFaultMessageList(ns6:ArrayOfFaultMessageList)
     ns6:FaultMessage(ns6:FaultMessage)
     ns6:FaultMessageList(ns6:FaultMessageList)
     ns4:CallContext(ns4:CallContext) …

我想知道是否有任何方法可以使用eclipse-pydev中的键盘快捷键自动修复所有PEP-8问题.谷歌搜索没有让我到任何地方.

由于Pydev可以检测到PEP-8问题,是否应该自动修复它们？

我在Eclipse Juno上使用Aptana studio插件.

我喜欢Aptana工作室提供的功能,但我不喜欢Aptana的查找和替换对话.

我想用Eclipse中的默认值替换它.

有可能,如果可能,怎么样？

所以我想试试jsonbPostgreSQL.在我的表,我有一个名为列extras的jsonb类型.

示例数据extras看起来像{"param1": 10, "param2": 15}

我想仅使用sql语句修改JSON.我想做这样的事情:

更新param1的extras通过,如果添加10到其值字段param2的extras超过12.

我该如何编写这样的SQL语句？我知道我可以很容易地在应用程序层中执行此操作,但我想在SQL层本身中执行此操作,因为我可能会处理的行数会很大而且我不想在db-application中浪费时间 - db往返

具体来说，MySQL的最大行大小限制为64kB。
我想知道Amazon Aurora是否存在相同的局限性，据说Amazon Aurora是MySQL的替代品。

我知道我能做到CREATE TABLE tbl_2 AS (select * from tbl_1)

但是有没有更好/更快/更强的方法来做到这一点？我现在谈论的主要是性能。这些表都是非规范化的，我没有任何外键约束需要担心。

编辑

或许就没有更好的办法了？参考：https://dba.stackexchange.com/questions/55661/how-to-duplicate-huge-postgres-table

我写了以下代码

$(function() {



    function get_updates () {
        $.getJSON('/new-lines.json', function(data) {
            var fullViewModel= ko.mapping.fromJS(data);
            ko.applyBindings(fullViewModel)


        });
    }

    function poll()
    {
        setTimeout(function(){
        get_updates();
        poll();},3000)

    }




    poll()
});

JSON数据如下所示:

{"state": "R", "qualities": ["ABC", "XYZ", "324"], "name": "ABC"}

我应该怎么写这个html部分？

我是javascript的新手.请帮忙.