-
Notifications
You must be signed in to change notification settings - Fork 1.1k
/
celery_app.py
63 lines (52 loc) · 1.87 KB
/
celery_app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
#########################################################################
#
# Copyright (C) 2016 OSGeo
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
#########################################################################
import os
import logging
from celery import Celery
# from celery.schedules import crontab
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "geonode.settings")
app = Celery("geonode")
logger = logging.getLogger(__name__)
def _log(msg, *args):
logger.debug(msg, *args)
# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()
app.autodiscover_tasks(packages=["geonode.harvesting.harvesters"])
""" CELERAY SAMPLE TASKS
@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
# Calls test('hello') every 10 seconds.
sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')
# Calls test('world') every 30 seconds
sender.add_periodic_task(30.0, test.s('world'), expires=10)
@app.task(
bind=True,
name='geonode.test',
queue='default')
def test(arg):
_log(arg)
@app.task(
bind=True,
name='geonode.debug_task',
queue='default')
def debug_task(self):
_log(f"Request: {self.request}")
"""