我正在我的django项目中编写测试.现在,我有两个数据库连接:
(settings.py)
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql_psycopg2',
'NAME': 'db_name'
...
},
}
和MongoDB的自定义连接:
import sys
from pymongo import Connection
from pymongo.errors import ConnectionFailure
try:
connection = Connection(host="localhost", port=27017)
db = connection['db_name']
print "Connected successfully(Mongo, db_name)"
except ConnectionFailure, e:
sys.stderr.write("Could not connect to MongoDB: %s" % e)
sys.exit(1)
我想知道我的项目何时开始运行
python manage.py test myapp
因为当您运行测试时,django会自动创建单独的DB(名称类似于test_db_name),但在这种情况下,Mongo仍将使用db_name运行.我试过了:
import sys
from pymongo import Connection
from pymongo.errors import ConnectionFailure
from django.db import connections
try:
connection = Connection(host="localhost", port=27017)
db_name = …默认情况下,Celery将所有任务发送到"celery"队列,但您可以通过添加额外参数来更改此行为:
@task(queue='celery_periodic')
def recalc_last_hour():
log.debug('sending new task')
recalc_hour.delay(datetime(2013, 1, 1, 2)) # for example
调度程序设置:
CELERYBEAT_SCHEDULE = {
'installer_recalc_hour': {
'task': 'stats.installer.tasks.recalc_last_hour',
'schedule': 15 # every 15 sec for test
},
}
CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler"
跑步工人:
python manage.py celery worker -c 1 -Q celery_periodic -B -E
此方案无法按预期工作:此工作人员将定期任务发送到"芹菜"队列,而不是"celery_periodic".我该如何解决这个问题？
PS芹菜== 3.0.16
我有计算纯python中相邻元素之间差异的算法:
a = range(1000000) #it's numpy array in my case
prev = a[0]
b = [0, ]
for i in a[1:]:
b.append(i - prev)
prev = i
有没有办法用Numpy重写这个功能？
在Celery中,您可以retry在发生异常时执行任何任务.你可以这样做:
@task(max_retries=5)
def div(a, b):
try:
return a / b
except ZeroDivisionError, exc:
raise div.retry(exc=exc)
在这种情况下,如果要除以零,任务将被重复五次.但是您必须明确检查代码中的错误.如果跳过try-except块,则不会重置任务.
我希望我的功能看起来像:
@celery.task(autoretry_on=ZeroDivisionError, max_retries=5)
def div(a, b):
return a / b
我有在纯python中计算平均速度的算法:
speed = [...]
avg_speed = 0.0
speed_count = 0
for i in speed:
if i > 0: # I dont need zeros
avg_speed += i
speed_count += 1
if speed_count == 0:
return 0.0
return avg_speed / speed_count
有没有办法用Numpy重写这个功能？
我想创建一个模型Changelog并使其可从管理页面进行编辑。以下是它的定义方式models.py：
class Changelog(models.Model):
id = models.AutoField(primary_key=True, auto_created=True)
title = models.TextField()
description = models.TextField()
link = models.TextField(null=True, blank=True)
picture = models.BinaryField(null=True, blank=True)
title和description是必需的，link并且picture是可选的。我想让这个模型尽可能简单，所以我选择BinaryField了FileField. 在这种情况下，我不需要担心需要备份的单独文件夹，因为 DB 将是独立的（我不需要存储文件名或任何其他属性，只需存储图像内容）。
我很快意识到，Django Admin 没有用于 的小部件BinaryField，因此我尝试将小部件用于FileField. 这是我为实现这一目标所做的（admin.py）：
class ChangelogForm(forms.ModelForm):
picture = forms.FileField(required=False)
def save(self, commit=True):
if self.cleaned_data.get('picture') is not None:
data = self.cleaned_data['picture'].file.read()
self.instance.picture = data
return self.instance
def save_m2m(self):
# FIXME: this function …在我的python代码中,我有全局requests.session实例:
import requests
session = requests.session()
我怎么能用它来嘲笑它Mock？这种操作有装饰器吗？我试过以下:
session.get = mock.Mock(side_effect=self.side_effects)
但是(正如预期的那样)这个代码session.get在每次测试后都没有返回到原始状态,就像@mock.patch装饰器一样.
我在python和C.中有几乎相同的代码.Python示例:
import numpy
nbr_values = 8192
n_iter = 100000
a = numpy.ones(nbr_values).astype(numpy.float32)
for i in range(n_iter):
a = numpy.sin(a)
C示例:
#include <stdio.h>
#include <math.h>
int main(void)
{
int i, j;
int nbr_values = 8192;
int n_iter = 100000;
double x;
for (j = 0; j < nbr_values; j++){
x = 1;
for (i=0; i<n_iter; i++)
x = sin(x);
}
return 0;
}
当我运行两个例子时,会发生奇怪的事情:
$ time python numpy_test.py
real 0m5.967s
user 0m5.932s
sys 0m0.012s
$ g++ sin.c
$ time ./a.out
real …我正在玩Linux系统调用,我发现了一些方面epoll,我不清楚.说,我创建一个epoll实例:
epollfd = epoll_create(50);
接下来,我在for-loop中注册了50个文件描述符:
for(i=0; i<50; i++){
// open file "file-i".txt
// construct epoll_event
// register new file descriptor with epoll_ctl(epollfd, EPOLL_CTL_ADD ...
现在我们有50个文件,已准备就绪(读取或写入 - 无关紧要).我们将MAX_EVENTS设置为3:
#define MAX_EVENTS 3
...
struct epoll_event events[MAX_EVENTS]
...
epoll_wait(epollfd, events, MAX_EVENTS, -1)
所有这50个文件都准备好了,我们只询问了其中的3个.哪些文件将在events数组中？
谢谢.
请考虑以下情况:您正在部署可以提供1 req./sec的应用程序.如果我在1秒内发送10个请求会怎么样？我写了一个简单的应用来测试:https://github.com/amezhenin/nginx_slow_upstream.此测试显示您的请求将被发送_in_exact_same_order_.
现在,这看起来像Nginx有一些请求队列,但我的同事(管理员)说Nginx中没有任何队列.所以我在这里写了一个关于epoll的问题:epoll是否保留了fd注册的顺序？.从那次讨论中我发现epoll确实保留了请求的顺序.
我有两个问题:
1)上面的推理/代码是否有任何错误？
2)Nginx是否在epoll之上有一些请求队列？或者Nginx使用纯epoll功能？
谢谢,抱歉我的英文:)