这篇文章主要讲解了“python测试开发django实例分析”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“python测试开发django实例分析”吧!
版本要求
Celery 5.2 版运行于
-
Python❨3.7、3.8、3.9、3.10❩
-
PyPy3.7、3.8 ❨7.3.7❩
Celery 4.x 是支持 Python 2.7 的最后一个版本,
Celery 5.x 需要 Python 3.6 或更高版本。
Celery 5.1.x 还需要 Python 3.6 或更高版本。
Celery 5.2.x 需要 Python 3.7 或更新版本。
如果您运行的是旧版本的 Python,则需要运行旧版本的 Celery:
Python 2.7 或 Python 3.5:Celery 系列 4.4 或更早版本。
Python 2.6:Celery 系列 3.1 或更早版本。
Python 2.5:Celery 系列 3.0 或更早版本。
Python 2.4 是 Celery 系列 2.2 或更早版本。
Celery 是一个资金很少的项目,所以不支持 Microsoft Windows。请不要打开与该平台相关的任何问题。
环境准备
运行系统:linux(centos/debian/ubuntu),不支持windows
Python版本:3.8.5
Django : 2.2.2
celery: 5.2.7
使用pip安装celery5.2.7版本
pip install celery==5.2.7
Django中使用Celery
要在 Django 项目中使用 Celery,您必须首先定义 Celery 库的实例(称为“应用程序”)
如果你有一个现代的 Django 项目布局,比如:
- proj/ - manage.py - proj/ - __init__.py - settings.py - urls.py
那么推荐的方法是创建一个新的proj/proj/celery.py模块来定义 Celery 实例:
proj/proj/celery.py
文件内容
import os from celery import Celery # Set the default Django settings module for the 'celery' program. os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings') app = Celery('proj') # Using a string here means the worker doesn't have to serialize # the configuration object to child processes. # - namespace='CELERY' means all celery-related configuration keys # should have a `CELERY_` prefix. app.config_from_object('django.conf:settings', namespace='CELERY') # Load task modules from all registered Django apps. app.autodiscover_tasks() @app.task(bind=True) def debug_task(self): print(f'Request: {self.request!r}')
然后你需要在你的proj/proj/init.py 模块中导入这个应用程序。这可以确保在 Django 启动时加载应用程序,以便@shared_task装饰器(稍后提到)将使用它:
proj/proj/__init__.py
内容:
# This will make sure the app is always imported when # Django starts so that shared_task will use this app. from .celery import app as celery_app __all__ = ('celery_app',)
请注意,此示例项目布局适用于较大的项目,对于简单的项目,您可以使用单个包含的模块来定义应用程序和任务
让我们分解第一个模块中发生的事情,首先,我们设置默认值DJANGO_SETTINGS_MODULEcelery命令行程序的环境变量:
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')
您不需要此行,但它使您不必总是将设置模块传递给celery程序。它必须始终在创建应用程序实例之前出现,就像我们接下来要做的那样:
app = Celery('proj')
这是我们的库实例,您可以有很多实例,但在使用 Django 时可能没有理由这样做。
我们还将 Django 设置模块添加为 Celery 的配置源。这意味着您不必使用多个配置文件,而是直接从 Django 设置中配置 Celery;但如果需要,您也可以将它们分开。
app.config_from_object('django.conf:settings', namespace='CELERY')
大写命名空间意味着所有 Celery 配置选项 必须以大写而不是小写指定,并且以 开头 CELERY_,例如task_always_eager设置变为CELERY_TASK_ALWAYS_EAGER,broker_url 设置变为CELERY_BROKER_URL。这也适用于工作人员设置,例如,worker_concurrency 设置变为CELERY_WORKER_CONCURRENCY.
例如,一个 Django 项目的配置文件可能包括:
... # Celery Configuration Options CELERY_TIMEZONE = "Australia/Tasmania" CELERY_TASK_TRACK_STARTED = True CELERY_TASK_TIME_LIMIT = 30 * 60
您可以直接传递设置对象,但使用字符串更好,因为这样工作人员不必序列化对象。CELERY_命名空间也是可选的,但建议使用(以防止与其他 Django 设置重叠)。
接下来,可重用应用程序的一个常见做法是在单独的tasks.py模块中定义所有任务,Celery 确实有一种方法可以自动发现这些模块:
app.autodiscover_tasks()
使用上面的代码,Celery 将自动从您安装的所有应用程序中发现任务,遵循tasks.py约定:
- app1/ - tasks.py - models.py - app2/ - tasks.py - models.py
这样您就不必手动将各个模块添加到CELERY_IMPORTS设置中。
最后,该debug_task示例是一个转储自己的请求信息的任务。这是使用bind=True Celery 3.1 中引入的新任务选项来轻松引用当前任务实例。
使用 @shared_task
装饰器
您编写的任务可能会存在于可重用的应用程序中,而可重用的应用程序不能依赖于项目本身,因此您也不能直接导入您的应用程序实例。
装饰器允许您在@shared_task没有任何具体应用实例的情况下创建任务:
demoapp/tasks.py:
# Create your tasks here from demoapp.models import Widget from celery import shared_task @shared_task def add(x, y): return x + y @shared_task def mul(x, y): return x * y @shared_task def xsum(numbers): return sum(numbers) @shared_task def count_widgets(): return Widget.objects.count() @shared_task def rename_widget(widget_id, name): w = Widget.objects.get(id=widget_id) w.name = name w.save()
django-celery-results 保存结果
django-celery-results- 使用 Django ORM/Cache 作为结果后端
django-celery-results扩展使用Django ORM 或 Django Cache 框架提供结果后端。
要将其用于您的项目,您需要执行以下步骤:
1.安装django-celery-results库:
pip install django-celery-results
2.添加django_celery_results到INSTALLED_APPS您的 Django 项目中settings.py:
INSTALLED_APPS = ( ..., 'django_celery_results', )
请注意,模块名称中没有破折号,只有下划线。
3.通过执行数据库迁移来创建 Celery 数据库表:
python manage.py migrate django_celery_results
4.配置 Celery 以使用django-celery-results后端。
假设您使用 Djangosettings.py来配置 Celery,添加以下设置
CELERY_RESULT_BACKEND = 'django-db'
对于缓存后端,您可以使用:
CELERY_CACHE_BACKEND = 'django-cache'
我们也可以使用 django 的 CACHES 设置中定义的缓存。
# celery setting. CELERY_CACHE_BACKEND = 'default' # django setting. CACHES = { 'default': { 'BACKEND': 'django.core.cache.backends.db.DatabaseCache', 'LOCATION': 'my_cache_table', } }
django-celery-beat 定时任务
django-celery-beat- 具有管理界面的数据库支持的定期任务。
启动工作进程
在生产环境中,您将希望在后台将工作程序作为守护程序运行 – 请参阅守护程序-但对于测试和开发,能够使用 celery worker manage 命令启动工作程序实例很有用,就像您一样d 使用 Django 的 manage.py runserver:
celery -A proj worker -l INFO
有关可用命令行选项的完整列表,请使用帮助命令:
celery help
感谢各位的阅读,以上就是“python测试开发django实例分析”的内容了,经过本文的学习后,相信大家对python测试开发django实例分析这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是云搜网,小编将为大家推送更多相关知识点的文章,欢迎关注!