小编Hus*_*ain的帖子

在芹菜中,我如何跟踪当前的重试？我知道我可以这样做:

@app.task(bind=True, default_retry_delay=900, max_retries=5)
def send_email(self, sender=None, to=None, subject=None, message=None):
  try:
    # Send email
  except Exception as e:
    # Print log message with current retry
    self.retry(e)

但我想说这是尝试1/5..2/5等等.

我已经从我的烧瓶式API模块中单独测试了所有方法.现在我想通过实际进行API调用来测试get方法.我期待这个测试的错误400.

我的资源类

class Response(Resource):

    @marshal_with(response_params_get_responses_on_job)
    def get(self, filter_name=None):
        try:
            response = self.process_get_request(filter_name)
            if not response['users']:
                raise MyValidationError("No data found")
            return response
        except MyValidationError as err:
            abort(404, message=err)
        except ValueError as mistake:
            abort(400, message=mistake)

我的单位测试

# TODO - Failing!
@mock.patch('application.resources.response.Response.process_get_request', autospec=True)
def test_get_400(self, process_get_request_mock):
    process_get_request_mock.side_effect = ValueError("some error")
    app = Flask(__name__)
    app.debug = True
    api = Api(app, prefix='/api/v2')
    api.add_resource(Response, '/user/responses', endpoint='job_responses')
    api.init_app(app)
    with app.test_client() as client:
        resp = client.get('/api/v2/user/responses',
                          environ_base={'HTTP_USER_AGENT': 'Chrome'},
                      headers={'Content-type': 'application/json'})
        self.assertEqual(resp.status_code, 400)

我的测试失败,因为我得到的响应是错误500

AssertionError: 500 != 400

堆栈跟踪 …

我有三个Celery任务，分别在三个不同的服务器上运行。

• task.send_push_notification
• task.send_sms
• task.send_email

我想设置一个工作流，以便如果发送推送通知失败，我应该尝试发送短信。如果发送短信失败，我应该发送电子邮件。

如果这3个任务及其代码库位于同一服务器上，那么我将按照链接任务的示例进行操作，例如

from celery import chain
from tasks import send_push_notification, send_sms, send_email
import json
# some paylaod
payload = json.dumps({})
res = chain(
    send_push_notification.subtask(payload),
    send_sms.subtask(payload),
    send_email.subtask(payload)
)()

但是任务被保存在3个不同的服务器上！

我试过了

# 1
from celery import chain
from my_celery_app import app
res = chain(
    app.send_task('tasks.send_push_notification', payload),
    app.send_task('tasks.send_sms', payload),
    app.send_task('tasks.send_email', payload)
)()
# Which fails because I am chaining tasks not subtasks

和

# 2
from celery import chain, subtask
res = chain(
    subtask('tasks.send_push_notification', payload), 
    subtask('tasks.send_sms', payload), …

我已经集成了 Salesforce 的 Bulk API，用于在我的 Python 项目中获取记录。带有'Content-Type': 'text/csv; charset=UTF-8'标题，它返回 CSV 输出。

"Id","Name","CreatedById","Salary","Base_Salary","Type","Pay_cycle","Description","Code"
"a0u90000003R4Y9AAK","Freelance Hadoop developer","005900000039GMdAAM","","","","","",""
"a0u90000003R5UPAA0","Senior Hadoop developer","005900000039GMdAAM","","","","","",""
"a0u90000003R5V3AAK","Freelance Webmaster","005900000039GMdAAM","","","","","",""
"a0u90000004ZMUwAAO","Full-Stack Web Developer - PHP and Python","005900000039GMdAAM","","","","","",""

现在，无需将其存储在 CSV 文件中，我想将这些记录放入我的数据库中。

这DictReader是获取字段值的一个非常好的和直接的解决方案，但我想它只适用于（CSV）文件。

import csv
with open('output.csv') as csvfile:
    reader = csv.DictReader(csvfile)
    for row in reader:
        print(row['Id'], row['Name'])

我正在尝试一些东西，但我不喜欢它。看

objects = []
for line in iter(csv_data.splitlines()):
    # Split by separator
    data = line.split(",")
    # Remove double quotes around the field value
    objects.append([i[1:-1] for i in data])

这样做的最优雅的方式是什么？

我有一个查询:

Update `pools` set status='2' where `pid`='1' and `uid`='2'

如何在cakephp中转换此查询？即我想在查询中传递AND条件,该查询应更新包含pid ='1'和uid ='2'的行.

我试图利用引导程序的切换按钮组来选择平台选项。但是该onChange事件没有被调用。

class Hello extends React.Component {
  constructor(props) {
    super(props);
    this.handlePlatformChange = this.handlePlatformChange.bind(this);
  }

  handlePlatformChange(event) {
    /* THIS DOES NOT GET CALLED */
    console.log(event.target.value);
  }

  render() {
    return (<div className="container">
      <div className="btn-group btn-group-toggle" data-toggle="buttons" onChange={this.handlePlatformChange} >
        <label className="btn btn-info active">
          <input type="radio" name="platform" value="web" autoComplete="off" /> Web
        </label>
        <label className="btn btn-info">
          <input type="radio" name="platform" value="android" autoComplete="off" /> Android
        </label>
        <label className="btn btn-info">
          <input type="radio" name="platform" value="ios" autoComplete="off" /> iOS
        </label>
      </div>
    </div>
    );
  }
}

我的应用程序正在使用webpack和以下依赖项

"dependencies": {
    "axios": …

我是django 1.10的新手,在Windows 8.1下使用mysql(5.6.17)(WAMP 2.5)和Python 3.5

我配置了很好的mysql连接器,我制作了我的模型,makemigrations myapp并migrate myapp使用带有Pycharm的manage.py终端.

当我尝试访问管理员网页时,我收到此消息(trabajo是我项目中的应用程序:programa)

例外价值:

(1146, "Table 'programa.django_session' doesn't exist")

在我的settings.py文件中有:

MIDDLEWARE_CLASSES = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
]

应用程序阻止:

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'trabajo',
]

要连接数据库:

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME' : 'programa',
        'USER' : 'myownuser',
        'PASSWORD' : 'myownpassword',
        'HOST' : '',
        'PORT' : '',
    }
}

为什么在管理页面中出现此错误？我该怎么做才能解决这个问题？

提前致谢.欢迎任何帮助!! ;-)

古斯塔沃.

启动,重启和停止的命令是什么gearman？

到目前为止,我知道:

gearadmin --shutdown # stop
gearmand -d # I believe it is to restart. Not sure.

管理的任何想法gearman？