小编dam*_*mon的帖子

每当我尝试运行命令时python manage.py syncdb,我都会收到以下错误:

Traceback (most recent call last):
  File "manage.py", line 11, in 
    execute_manager(settings)
  File "/home/damon/Workspace/django-projects/acm-cie/env/lib/python2.6/site-packages/django/core/management/__init__.py", line 438, in execute_manager
    utility.execute()
  File "/home/damon/Workspace/django-projects/acm-cie/env/lib/python2.6/site-packages/django/core/management/__init__.py", line 379, in execute
    self.fetch_command(subcommand).run_from_argv(self.argv)
  File "/home/damon/Workspace/django-projects/acm-cie/env/lib/python2.6/site-packages/django/core/management/__init__.py", line 261, in fetch_command
    klass = load_command_class(app_name, subcommand)
  File "/home/damon/Workspace/django-projects/acm-cie/env/lib/python2.6/site-packages/django/core/management/__init__.py", line 67, in load_command_class
    module = import_module('%s.management.commands.%s' % (app_name, name))
  File "/home/damon/Workspace/django-projects/acm-cie/env/lib/python2.6/site-packages/django/utils/importlib.py", line 35, in import_module
    __import__(name)
  File "/home/damon/Workspace/django-projects/acm-cie/env/lib/python2.6/site-packages/south/management/commands/__init__.py", line 10, in 
    import django.template.loaders.app_directories
  File "/home/damon/Workspace/django-projects/acm-cie/env/lib/python2.6/site-packages/django/template/loaders/app_directories.py", line 21, in 
    mod = import_module(app)
  File "/home/damon/Workspace/django-projects/acm-cie/env/lib/python2.6/site-packages/django/utils/importlib.py", line 35, in import_module …

如果我将一个环境变量作为参数传递给容器内部docker run,我的shell会对其进行求值.例如:

我希望我的容器打印出的值$FOO是bar.这些都不会起作用:

# Prints blank line
$ docker run -e FOO=bar ubuntu echo $FOO

# Prints '$FOO'
$ docker run -r FOO=bar ubuntu echo \$FOO

我使用的是django 2.0.8和Python 3.5.我希望能够在将对象保存到数据库时发送和接收自定义信号.

我已经关注了Django关于收听信号的文档以及与Django捆绑在一起的核心信号 - 但是,我无法让我的示例工作.

这是我到目前为止:

的myapp/models.py

from django.db import models
import django.dispatch

my_signal = django.dispatch.Signal(providing_args=["name"])

class Foo(models.Model):
    name = models.CharField(max_length=16)

def save(self, *args, **kwargs):
    try:
        # Call the "real" save() method.
        super().save(*args, **kwargs)  

        # Fire off signal         
        my_signal.send(sender=self.__class__, name=self.name)

    except Exception as e:
        print ('Exception:', e)
        #pass

MYAPP/signals.py

from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Foo


@receiver(post_save, sender=Foo)
def foo_handler(sender, **kwargs):
    print('Foo Signal Recieved!')
    print(kwargs)

MYAPP/app.py

class MyappConfig(AppConfig):
    name …

我希望使用 AWS Secrets Manager 来获取机密并将它们设置为我的 Elastic Beanstalk 实例上的环境变量。

我已经在 ebextensions 文件上编写了一个脚本，该脚本调用 Secrets Manager CLI 来获取我的秘密，并使用该秘密来填充我的 EB 实例的 env 变量。由于它是 linux 实例，我正在尝试export ENV_VAR_NAME=env_value. 这是我到目前为止所拥有的：

packages:
  yum:
    epel-release: []
    jq: []

files:
  "/home/ec2-user/test.sh" :
    mode: "000755"
    owner: root
    group: root
    content: |
      #!/bin/bash
      config=$(aws --region us-west-1 secretsmanager get-secret-value --secret-id secret | jq -rc '.SecretString'')

      export SECRET_KEY=$(echo $config | jq -rc '.awsKey')

      # Used to print current env variables
      env

commands:
  0_test:
    command: /home/ec2-user/test.sh

#I've also tried replacing 'commands' with 'container_commands' …

我正在失去理智，试图找到一种可靠且可测试的方法来获取给定 Celery 队列中包含的任务数量。

我已经阅读了这两个相关的讨论：

• Django Celery 获取任务计数

  注意：我没有使用 Django 或任何其他 Python Web 框架。

• 检索 Celery 队列中的任务列表

但我无法使用这些线程中描述的方法解决我的问题。

我使用 Redis 作为后端，但我希望有一个独立于后端且灵活的解决方案，特别是对于测试。

这是我目前的情况：我定义了一个EnhancedCelery类，它继承Celery并添加了几个方法，特别get_queue_size()是我正在尝试正确实现/测试的方法。

以下是我的测试用例中的代码：

celery_test_app = EnhancedCelery(__name__)

# this is needed to avoid exception for ping command
# which is automatically triggered by the worker once started
celery_test_app.loader.import_module('celery.contrib.testing.tasks')

# in memory backend
celery_test_app.conf.broker_url = 'memory://'
celery_test_app.conf.result_backend = 'cache+memory://'

# We have to setup queues manually, 
# since it seems that auto queue creation doesn't work in …

我开始在我的django Web应用程序中使用AWS Elasticache.

我首先使用自动发现功能将缓存位置设置为唯一端点,但似乎不起作用.

我正在使用pylibmc(1.2.2)和django-pylibmc-sasl(0.2.4)连接到python的memcached.

自动发现功能是否适用于这些客户端？我该如何启用它？

我试图理解assert_called_with内部模拟，但我编写的代码引发了一些错误。

import os
import twitter

URL = "http://test.com"

def tweet(api, message):
    if len(message) > 40:
        message = message.strip("?.,.,.")

    status = api.PostUpdate(message)
    return status

def main():
    api = twitter.Api(consumer_key=''
                    ,consumer_secret='')
    msg = 'This is test message'
    tweet(api, msg)

if __name__ == '__main__':
    main()

单元测试

import unittest
from mock import Mock
import test

class TweetTest(unittest.TestCase):
    def test_example(self):
        mock_twitter = Mock()        
        test.tweet(mock_twitter,'msg')        
        mock_twitter.PostUpdate.assert_called_with('message')        

if __name__ == '__main__':
    unittest.main()

我想了解assert_called_with这里有什么？

我有一个关于 for 信号的使用的问题dispatch_uid。

目前，我通过简单地添加if not instance.order_reference. 我现在想知道是否dispatch_uid具有相同的功能，并且我可以删除“ if not”子句。

信号.py

def reserveditem_create_order_reference(sender, instance, **kwargs):
    if not instance.order_reference:
        instance.order_reference = unique_order_reference_generator()

应用程序.py

class OrdersConfig(AppConfig):
    name = 'orders'

    def ready(self):

        #Pre save signal for ReservedItem model
        reserved_model = self.get_model('ReservedItem')
        pre_save.connect(
            reserveditem_create_order_reference,
            sender=reserved_model,
            dispatch_uid="my_unique_identifier"
        )

I have created a celery worker with a single celerybeat schedule task which runs at 5 seconds time interval. How can I add another beat task dynamically to the celery worker without stopping it?

Example

app.conf.update(
   CELERY_TASK_RESULT_EXPIRES=3600,
   CELERY_TIMEZONE = 'UTC',
   CELERYBEAT_SCHEDULE = {
    'long-run-5-secs': {
        'task': 'test_proj.tasks.test',
        'schedule': timedelta(seconds=5),
        'args': (16, )
    }
   }
)

With the above configuration, I am able to run the celery worker with beat mode successfully.

Now I need add the below beat schedule dynamically: …

我有以下型号:

class Indicator(models.Model):
    name = models.CharField(max_length=200)
    category = models.ForeignKey(IndicatorCategory)
    weight = models.IntegerField()
    industry = models.ForeignKey(Industry)

    def __unicode__(self):
        return self.name
    class Meta:
        ordering = ('name',)

class IndicatorRatingOption(models.Model):
    indicator = models.ForeignKey(Indicator)
    description = models.TextField()
    value = models.FloatField(null=True)

    def __unicode__(self):
        return self.description

class Rating(models.Model):
    product = models.ForeignKey(Product, null=True)
    company = models.ForeignKey(Company, null=True)
    rating_option = models.ForeignKey(IndicatorRatingOption)
    value = models.IntegerField(null=True)

我需要做的是获得两家公司的所有公司评级选项,而不会在指标(rating.rating_option.indicator)上重叠.如果存在冲突,公司'a'将永远战胜公司'b'.我该怎么做呢？