在django中使用apscheduler 执行计划任务的实现方法

时间：2021-01-14 11:14:08|栏目：Python代码|点击：次

对于任何软件开发人员而言，为将来计划任务都是必不可少的工具。尽管我们创建的许多编程旨在响应明确的触发或用户事件，但定期执行的后台进程也同样重要。

“每个星期一早晨更新结果。”

“每天晚上分批下单。”

甚至具有每日请求限制的第三方API也隐式要求这种行为。

“我们只能每五分钟请求一次更新。”

幸运的是，许多聪明的人已经解决了这个问题，并且不难找到python本地解决方案。 Advanced Python Scheduler（APS）是一个很好的选择，它具有简单，直观的API以及同类产品中的一些最佳文档。

对于此项目，我们将专注于将APS提供的调度技术与您的常规Django应用程序集成：洛杉矶天气应用程序，该应用程序定期轮询第三方天气api以进行模型更新。

目标是比Django教程进行更深入的探索，同时不要在任何方向上陷入困境。

I.安装APS和其他依赖项

在您的项目目录中，创建一个虚拟环境并激活它

virtualenv env
. env/bin/activate

根据本指南安装和配置PostgreSQL。在此阶段，我们只需要在您的计算机上启动并运行SQL管理器即可。

另外，我发现使用PgAdmin PostgreSQL GUI有帮助。在您的计算机上进行设置的详细信息可以在这里找到（使用Python3）。

使用pip安装所有必需的软件包（注意，psycopg2适用于PostgreSQL）：

pip install apscheduler django psycopg2 requests
II. Build your app
Create a new Django project:

django-admin.py startproject advancedScheduler
cd advancedScheduler
python manage.py startapp weather

在这个新目录（根目录）中，您将看到另一个名为advancedScheduler的文件夹。这是Django项目目录。

为避免两地同名的混淆，我们仅将“根目录”称为“根目录”。让下面的图作为我们跳文件夹冒险的路线图。

[ super_project_directory/ ]
|
+----[ env/ ] <-- Virtualenv stuff
|
+----[ advancedScheduler/ ] <-- the Root Directory
 |
 +----[ advancedScheduler/ ] <-- the Django Project Directory
 |
 +----[ weather/ ] <-- the Django App Directory

尽管主要专注于演示调度程序的功能，但让我们花点时间连接Django应用。

我们首先要将天气应用添加到项目的INSTALLED_APPS中。该文件位于advancedScheduler / settings.py文件中。

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

接下来，将新的网址格式添加到advancedScheduler / urls.py文件中：

path('', include('weather.urls'))

毫不奇怪，我们的下一步将是将该urls.py文件添加到weather app目录中。将以下代码包含在weather / urls.py中：

from django.conf.urls import url
from weather import views
urlpatterns = [
 url(r'^$', views.MainPage.as_view())
]

在天气应用程序目录中创建一个模板文件夹。将index.html文件添加到此新文件夹。

以下是我们的MTV。

模型

from django.db import models
from datetime import datetime 
 
class Forecast(models.Model):
 timestamp = models.DateTimeField()
 temperatue = models.DecimalField(max_digits=12,decimal_places=2)
 description = models.CharField(max_length=150)
 city = models.CharField(max_length=150)
 
 def save(self, *args, **kwargs):
 if not self.id:
  self.timestamp = datetime.utcnow()
 return super(Forecast, self).save(*args, **kwargs)
Template
<div style="text-align: center">
 <h5>Currently in</h5>
 <h3>{{city}}</h3>
 <h4>{{temperature_in_f}} F | {{temperature_in_c}} C</h4>
 <h4>{{desctiprion}}</h4>
 <p><em>Last updated {{utc_update_time}} GMT</em></p>
</div>
View
import decimal
from datetime import datetime 
from django.shortcuts import render
from django.views.generic import TemplateView
from weather.models import Forecast 
class MainPage(TemplateView):
 def get(self, request, **kwargs):
 
 latest_forecast = Forecast.objects.latest('timestamp')
 city = latest_forecast.city
 temperature_in_c = latest_forecast.temperatue
 temperature_in_f = (latest_forecast.temperatue * decimal.Decimal(1.8)) + 32
 description = latest_forecast.description.capitalize
 timestamp = "{t.year}/{t.month:02d}/{t.day:02d} - {t.hour:02d}:{t.minute:02d}:{t.second:02d}".format( t=latest_forecast.timestamp)
 return render(
  request, 
  'index.html', 
  {
  'city':city,
  'temperature_in_c': temperature_in_c,
  'temperature_in_f': round(temperature_in_f,2),
  'desctiprion': description,
  'utc_update_time': timestamp})

三，建立数据库连接并迁移模型

在advancedScheduler / settings.py中，将DATABASES值更改为：

DATABASES = {
 'default': {
 'ENGINE': 'django.db.backends.postgresql_psycopg2',
 'NAME': 'advancedScheduler',
 'USER': 'some_user_name',
 'PASSWORD': 'some_password',
 'HOST': 'localhost',
 'PORT': '',
 }
}

您应该从上述PostgreSQL配置指南（此处和此处）了解USER，PASSWORD和PORT的值。

与PostgreSQL建立连接后，就该迁移我们的模型了。导航到“根目录”并键入：

python manage.py makemigrations
python manage.py migrate

这样，我们的模型应该已经映射到数据库了。继续并检查所有内容。不用担心，我会在这里等你回来。

IV。预测API

时间到了有趣的部分。我正在从OpenWeatherMap（一个免费的天气API）中提取我的预报数据，该API将为您授予带有有效电子邮件地址的访问令牌。

现在，由于它在概念上不同于我们的表示层，因此让我们在根目录中创建一个新的ForecastUpdater文件夹。在其中，我们将添加两个文件：一个空白的__init__.py文件和一个ForecastApi.py文件。请参阅路线图以供参考。

[ super_project_directory/ ]
|
+----[ env/ ]
|
+----[ advancedScheduler/ ] <-- the Root Directory
 |
 +----[ advancedScheduler/ ] 
 |
 +----[ weather/ ] 
 |
 +----[ forecastUpdater/ ] <-- the new Updater Module
  |
  +----< __init__.py > <--+
  |    |-- two new Python files
  +----< forecastApi.py > <--+
import requests
from weather.models import Forecast
 
def _get_forecast_json():
 url = 'http://api.openweathermap.org/data/2.5/weather'
 encoded_city_name = 'Los%20Angeles'
 country_code = 'us'
 access_token = 'your_access_token'
 
 r = requests.get('{0}?q={1},{2}&APPID={3}'.format(
 url, 
 encoded_city_name, 
 country_code, 
 access_token))
 
 try:
 r.raise_for_status()
 return r.json()
 except:
 return None
 
 
def update_forecast():
 json = _get_forecast_json()
 if json is not None:
 try:
  new_forecast = Forecast()
  
  # open weather map gives temps in Kelvin. We want celsius.  
  temp_in_celsius = json['main']['temp'] - 273.15
  new_forecast.temperatue = temp_in_celsius
  new_forecast.description = json['weather'][0]['description']
  new_forecast.city = json['name']
  new_forecast.save()
  print("saving...\n" + new_forecast)
 except:
  pass

在这里，有一些事情要注意。异常处理远非健壮。错误只是被丢弃了―过度的沉默是唯一出问题的迹象。

其次，我们在代码中指定洛杉矶。将您的服务器配置到所需的任何位置。

同样重要的是要注意，update_forecast（）不带任何参数。我们很快就会看到，我们的高级python计划程序具有严格的无参数规则。甚至带有孤独的self参数的方法也不会飞。

五，高级Python计划程序

我们已经建立了模型。我们可以通过调用API来更新数据。现在我们需要做的就是指定访问该API的频率，这样我们就可以在不超出数据访问限制的情况下提供合理的最新信息。

在ForecastUpdater模块中，添加一个updater.py文件。在这里，我们将使用Advanced Python Scheduler设置我们的预测更新的节奏。

OpenWeatherMaps使用条款允许在一个小时内保持60个通话，以保持免费等级；每五分钟更新一次就足够了。

from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
from forecastUpdater import forecastApi
 
def start():
 scheduler = BackgroundScheduler()
 scheduler.add_job(forecastApi.update_forecast, 'interval', minutes=5)
 scheduler.start()

这可能是您可以找到的最简单的APS实现。如果您查看他们的网站或GitHub上的几个工作示例，则将发现一个完整的功能和设置工具箱，您可以使用这些工具来进行计时，以使其尽可能的细致。

按照我们想要的方式配置了调度程序后，就可以将其连接到Django应用了。

理想情况下，我们希望在调度程序上按一次播放，然后让它执行其任务。我们需要一种一致且可靠的方式来初始化时间表一次且仅一次。对于我们而言，Django正是这种类型的运行时初始化逻辑的地方。

在weather / apps.py文件中，您会找到一个名为WeatherConfig的类的存根，该类继承自Django的AppConfig类。

class WeatherConfig(AppConfig):
 name = 'weather'

为了让Django知道它需要在启动时启动更新程序，我们覆盖了AppConfig.ready（）方法。

from django.apps import AppConfig
class WeatherConfig(AppConfig):
 name = 'weather'
 def ready(self):
 from forecastUpdater import updater
 updater.start()

重要的是要记住，由于继承的复杂性，此覆盖的任何导入都必须位于ready（）方法的主体内。 Django还警告不要在我们的覆盖中直接与数据库进行交互；生产，调试，风雨无阻，每次启动天气应用程序时，都会执行此代码。

最后，我们现在需要在advancedScheduler / settings.py中再次更新INSTALLED_APPS变量。 Django需要知道我们要使用自定义配置来运行天气应用。

INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'weather.apps.WeatherConfig' 
]

VI。全部放在一起

而已。在这一点上，我们可以启动我们的应用程序，然后让更新程序执行其操作。

python manage.py runserver --noreload

-noreload标志可防止Django启动天气应用的第二个实例-这是调试模式下的默认行为。第二个实例意味着我们所有计划的任务将触发两次。

最初，我们的结果看起来不完整。由于我们将更新程序逻辑安排为每五分钟运行一次，因此我们不停地抽动一下……为了使事情变得有趣，缩短审慎刷新之间的间隔可能是明智的选择，或者在初始化时调用一次update_forecast（）。

七。最后的想法

我们做到了！我们的天气应用已准备好与世界分享（请在此处查看我的信息）。

Advanced Python Scheduler是任何Python开发人员都知道的好工具。它在直观的API后面隐藏了非常常见的业务需求的复杂性。考虑一下，安装程序只用了三行代码。

该项目的真正技巧是与Django框架进行交互-配置，迁移，初始化。然后，任务自动化成为事后的想法。五分钟内您就完成了。

https://github.com/kmhoran/la-weather-app

以上所述是小编给大家介绍的在django中使用apscheduler 执行计划任务的实现方法，希望对大家有所帮助！

上一篇：opencv resize图片为正方形尺寸的实现方法