celery: how to separate different environments with different workers?

CELERY_BROKER = "librabbitmq://user:user@my_machine.domain.com/"

CELERY_BROKER = "librabbitmq://prop:prod@main_cluster.domain.com/"

app = Celery('App', broker = settings.CELERY_BROKER)

CELERY_BROKER = "librabbitmq://user:user@my_machine.domain.com/"

CELERY_BROKER = "librabbitmq://prop:prod@main_cluster.domain.com/"

app = Celery('App', broker = settings.CELERY_BROKER)

$ heroku apps: create
$ heroku addons: create heroku - redis - a sushi

$ pip install redis

$ pip install celery

@app.task
def add(x, y):
   return x + y

import os
app.conf.update(BROKER_URL = os.environ['REDIS_URL'],
   CELERY_RESULT_BACKEND = os.environ['REDIS_URL'])

transport: //userid:[email protected]:port/virtual_host

#!/bin/sh

if [$1 = "beat"];
then
celery - A path.to.celery.app beat--loglevel = INFO
else
   celery - A path.to.celery.app worker--concurrency = 4--loglevel = INFO--without - gossip--without - mingle--without - heartbeat - Ofair
fi

services:

   web: [...]

database_default: [...]

rabbitmq:
   image: rabbitmq: 3.9 - management
hostname: rabbitmq
ports:
   -"15672:15672"
expose:
   -"15672"

celeryworker:
build: "."
links:
- "database_default"
- "rabbitmq:rabbitmq"
volumes:
- ".:/app:rw"
- "./data:/data:rw"
command: <startup command>
   env_file: .env-local

celerybeat:
build: "."
links:
- "database_default"
- "rabbitmq:rabbitmq"
volumes:
- ".:/app:rw"
- "./data:/data:rw"
command: <startup command>
   env_file: .env-local

celerycam:
   build: "."
links:
   -"database_default" -
   "rabbitmq:rabbitmq"
volumes:
   -".:/app:rw" -
   "./data:/data:rw"
command: aldryn - celery cam
env_file: .env - local

# # Broker settings.
BROKER_URL = 'amqp://guest:guest@localhost:5672//'

# List of modules to
import when celery starts.
CELERY_IMPORTS = ('myapp.tasks', )

# # Using the database to store task state and results.
CELERY_RESULT_BACKEND = 'db+sqlite:///results.db'

CELERY_ANNOTATIONS = {
   'tasks.add': {
      'rate_limit': '10/s'
   }
}

CELERY_ANNOTATIONS = {
   'tasks.add': {
      'rate_limit': '10/s'
   }
}

CELERY_ANNOTATIONS = {
   '*': {
      'rate_limit': '10/s'
   }
}

def my_on_failure(self, exc, task_id, args, kwargs, einfo):
   print('Oh no! Task failed: {0!r}'.format(exc))

CELERY_ANNOTATIONS = {
   '*': {
      'on_failure': my_on_failure
   }
}

class MyAnnotate(object):

   def annotate(self, task):
   if task.name.startswith('tasks.'):
   return {
      'rate_limit': '10/s'
   }

CELERY_ANNOTATIONS = (MyAnnotate(), {
   …})

CELERY_RESULT_BACKEND = 'db+scheme://user:password@host:port/dbname'

    -celery_uncovered /
       -celery_uncovered / __init__.py -
       celery_uncovered / {
          toyex,
          tricks,
          advex
       } -
       celery_uncovered / celery.py -
       config / settings / {
          base,
          local,
          test
       }.py -
       config / urls.py -
       manage.py

from __future__
import absolute_import

import os
from celery
import Celery, signals

# set the
default Django settings module
for the 'celery'
program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.local')

app = Celery('celery_uncovered')

# 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()

from __future__
import absolute_import

# 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 # noqa

__all__ = ['celery_app']

__version__ = '0.0.1'
__version_info__ = tuple([int(num) if num.isdigit()
   else num
   for num in __version__.replace('-', '.', 1).split('.')
])

@shared_task
def fetch_hot_repos(since, per_page, page):
   payload = {
      'sort': 'stars',
      'order': 'desc',
      'q': 'created:>={date}'.format(date = since),
      'per_page': per_page,
      'page': page,
      'access_token': settings.GITHUB_OAUTH
   }
headers = {
   'Accept': 'application/vnd.github.v3+json'
}
connect_timeout, read_timeout = 5.0, 30.0
r = requests.get(
   'https://api.github.com/search/repositories',
   params = payload,
   headers = headers,
   timeout = (connect_timeout, read_timeout))
items = r.json()[u 'items']
return items