python类Celery()的实例源码

def CreateCelery(app = None):
    app = app or CreateApp()
    if not NodeDefender.config.celery.enabled():
        NodeDefender.logger.info("Concurrency disabled")
        return False

    try:
        celery = Celery(app.name, broker=app.config['CELERY_BROKER_URI'],
                   backend=app.config['CELERY_BACKEND_URI'])
    except KeyError:
        celery = Celery(app.name)
        NodeDefender.logger.info("Concurreny configuration error")
        return False

    celery.conf.update(app.config)
    TaskBase = celery.Task
    class ContextTask(TaskBase):
        abstract = True
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    celery.Task = ContextTask
    NodeDefender.logger.info("Concurrency successfully enabled")
    return celery

def make_celery(app):
    """
    desc:   ??celery?flask?

    """
    celery = Celery(app.import_name, broker=app.config["CELERY_BROKER_URL"],
                    backend=app.config["CELERY_BACKEND_URL"])
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask
    return celery

def make_celery(app):
    """Create a celery instance for the application."""
    # Celery is optional, import it here rather than globally
    from celery import Celery

    celery_instance = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )

    celery_instance.conf.update(app.config)
    TaskBase = celery_instance.Task

    class ContextTask(TaskBase):
        abstract = True
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery_instance.Task = ContextTask

    return celery_instance

def celery(self):
        app = self.flask_app
        celery = Celery(
                app.import_name,
                broker=app.config['CELERY_BROKER_URL'])

        celery.conf.update(app.config)

        TaskBase = celery.Task
        class ContextTask(TaskBase):
            abstract = True

            def _call_(self, *args, **kwargs):
                with app.app_context():
                    return TaskBase.__call__(self, *args, **kwargs)

        celery.Task = ContextTask

        return celery

def make_celery(app):
    celery = Celery(app.import_name,
                    backend=app.config['CELERY_RESULT_BACKEND'],
                    broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask
    return celery


# Instantiate Flask app

def create_celery(name, config_obj, inject_version=True, **kwargs):
    """Creates a celery app.

    :param config_obj: The configuration object to initiaze with
    :param inject_version: bool: Whether or not to inject the application's
                                 version number. Attempts to get version number
                                using 
                                :func:`twopi_flask_utils.deployment_release.get_release`
    :param kwargs: Other arguments to pass to the ``Celery`` instantiation.
    :returns: An initialized celery application.
    """

    celery = Celery(name, broker=config_obj.CELERY_BROKER_URL, **kwargs)
    celery.config_from_object(config_obj)
    if inject_version:
        celery.version = get_release()

    return celery

def flask_celery(flask_apm_client):
    from celery import Celery

    flask_app = flask_apm_client.app
    celery = Celery(flask_app.import_name, backend=None,
                    broker=None)
    celery.conf.update(CELERY_ALWAYS_EAGER=True)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with flask_app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask
    celery.flask_apm_client = flask_apm_client
    return celery

def test_celery(self):
        # get a Celery "connection"
        celery = Celery()

        # This is a test task that returns the number
        # passed into the function.
        @celery.task
        def mirror(x):
            return x

        # some number.
        test_n = 44
        # kick off the celery task
        r = mirror.apply(args=(test_n,)).get()
        # ensure the number comes back.
        self.assertEqual(r, test_n)

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config_type[config_name])
    config_type[config_name].init_app(app)

    celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)

    db.init_app(app)
    mail.init_app(app)
    bcrypt.init_app(app)
    login_manager.init_app(app)

    from .main import main as main_blueprint
    app.register_blueprint(main_blueprint)

    return app

def make_celery(app):
    celery = Celery(
        app.import_name,
        broker=app.config['CELERY_BROKER_URL'],
        backend=app.config['CELERY_BACKEND_URL']
    )
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask

    return celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        broker=app.config['CELERY_BROKER_URL'],
        backend=app.config['CELERY_BACKEND_URL']
    )
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask

    return celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        broker=app.config['CELERY_BROKER_URL'],
        backend=app.config['CELERY_BACKEND_URL']
    )
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask

    return celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        broker=app.config['CELERY_BROKER_URL'],
        backend=app.config['CELERY_BACKEND_URL']
    )
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask

    return celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        broker=app.config['CELERY_BROKER_URL'],
        backend=app.config['CELERY_BACKEND_URL']
    )
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask

    return celery

def make_celery(app):
    celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    celery.Task = ContextTask
    return celery



# Making SqlAlchemy instance of the app

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['NIXBORG_CELERY_RESULT_BACKEND'],
        broker=app.config['NIXBORG_CELERY_BROKER_URL'],
    )
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask

    return celery

def make_celery(app):
    celery=Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )

    celery.conf.update(app.config)
    TaskBase=celery.Task

    class ContextTask(TaskBase):
        abstract=True
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self,*args,**kwargs)

    celery.Task=ContextTask
    return celery

def make_celery(app):
    """
    :function: make_celery
    :args:
        - app: restccnu app
    :rv: celery??

    celery????, ?celery context???flask app context,
    ????flask??
    """
    celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    TaskBase = celery.Task
    class ContextTask(TaskBase):
        abstract = True  # abc
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    celery.Task = ContextTask
    return celery

def create_celery_app(app=None):
    """Return a celery app in app context"""
    app = app or create_app(
        os.environ.get("ENV", "prod"), register_blueprints=False)
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL'])

    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask
    return celery

def test_task():
    from pylogctx import context

    app = Celery(task_cls=LoggingTask)

    @app.task
    def my_task():
        context.update(taskField='RUNNED')
        logger = get_task_logger(current_task.name)
        logger.info("I log!")
        return context.as_dict()

    result = my_task.apply()
    if VERSION.major < 4:
        result.maybe_reraise()
    else:
        result.maybe_throw()
    fields = result.result
    assert 'taskField' in fields
    assert not context.as_dict()

def test_adapter():
    from pylogctx import context, log_adapter

    app = Celery(task_cls='pylogctx.celery.LoggingTask')

    @log_adapter(app.Task)
    def adapter(task):
        return {
            'celeryTaskId': task.request.id,
            'celeryTask': task.name
        }

    @app.task
    def my_task():
        return context.as_dict()

    result = my_task.apply()
    if VERSION.major < 4:
        result.maybe_reraise()
    else:
        result.maybe_throw()

    fields = result.result
    assert 'celeryTask' in fields
    assert 'celeryTaskId' in fields

def init_app(self, app, sentry):
        self.app = app
        new_celery = celery.Celery(
            app.import_name,
            broker=app.config['CELERY_BROKER_URL'],
            backend=app.config['CELERY_RESULT_BACKEND'],
        )
        # XXX(dcramer): why the hell am I wasting time trying to make Celery work?
        self.celery.__dict__.update(vars(new_celery))
        self.celery.conf.update(app.config)

        worker_process_init.connect(self._worker_process_init)

        task_postrun.connect(self._task_postrun)
        task_prerun.connect(self._task_prerun)

        if sentry:
            register_signal(sentry.client)
            register_logger_signal(sentry.client)

def parse_celery_config(celery_config_python: str) -> dict:
    # Expose timedelta object for config to be used in beat schedule
    # http://docs.celeryproject.org/en/master/userguide/periodic-tasks.html#beat-entries
    from datetime import timedelta  # noqa
    from celery.schedules import crontab  # noqa

    _globals = globals().copy()
    _locals = locals().copy()
    code = textwrap.dedent(celery_config_python)

    try:
        config_dict = eval(code, _globals, _locals)
    except Exception as e:
        raise RuntimeError("Could not execute Python code to produce Celery configuration object: {}".format(code)) from e

    if "broker_url" not in config_dict:
        raise RuntimeError("Mandatory broker_url Celery setting missing. Did we fail to parse config? {}".format(config_dict))

    return config_dict

def get_celery_config(registry: Registry) -> dict:
    """Load Celery configuration from settings.

    You need to have a setting key ``celery_config_python``. This is Python code to configure Celery. The code is executed and all locals are passed to Celery app.

    More information:

        * http://docs.celeryproject.org/en/master/userguide/configuration.html

    :param registry: Pyramid registry from where we read the Celery configuratino
    :return: An object holding Celery configuration variables
    """

    celery_config_python = registry.settings.get('websauna.celery_config')
    if not celery_config_python:
        raise RuntimeError('Using Celery with Websauna requires you to have celery_config_python configuration variable')

    return parse_celery_config(celery_config_python)

def get_celery(registry: Registry):
    """Load and configure Celery app.

    Cache the loaded Celery app object on registry.

    :param registry: Use registry settings to load Celery
    """
    celery = getattr(registry, "celery", None)
    if not celery:
        celery = registry.celery = Celery()
        celery.conf.update(get_celery_config(registry))

        # Expose Pyramid registry to Celery app and tasks
        celery.registry = registry

    return celery

def update(self, schedule):
        """Update the registered schedule by reconciliating with database
        backend.

        :param schedule: Schedule to synchronize with the database.
        :type schedule: :class:`~news.models.AbstractSchedule` implementation

        """
        if isinstance(schedule, int):
            schedule = self.backend.get_schedule_by_id(schedule)

        # log
        self._log('Updating schedule {}'.format(
            schedule if isinstance(schedule, int) else schedule.id))

        # remove schedule from job queue and add it if it's now enabled
        self.remove_schedule(schedule)
        if schedule.enabled:
            self.add_schedule(schedule)

    # ==================
    # Celery integration
    # ==================

def init_celery(app=None, result_backend=True):
    """Init Celery configuration.

    :param app: celery configuration, if omitted, application will be instantiated
    :param result_backend: True if Celery should connect to result backend
    """
    # Keep this for debugging purposes for now
    _logger.debug(">>> Selinon version is %s" % selinon_version)
    _logger.debug(">>> Selinonlib version is %s" % selinonlib_version)
    _logger.debug(">>> Celery version is %s" % celery_version)

    if not result_backend:
        CelerySettings.disable_result_backend()

    if app is None:
        app = Celery('tasks')
        app.config_from_object(CelerySettings)
    else:
        app.config_from_object(CelerySettings)

def create_celery_app(app=None):
  app = app or create_app()
  celery = Celery(__name__, broker=app.config['CELERY_BROKER_URL'])
  celery.conf.update(app.config)
  TaskBase = celery.Task

  class ContextTask(TaskBase):
    abstract = True

    def __call__(self, *args, **kwargs):
      with app.app_context():
        return TaskBase.__call__(self, *args, **kwargs)

  celery.Task = ContextTask
  celery.app = app
  return celery

def ready(self):
        # Using a string here means the worker will not have to
        # pickle the object when using Windows.
        app.config_from_object('django.conf:settings')
        installed_apps = [app_config.name for app_config in apps.get_app_configs()]
        app.autodiscover_tasks(lambda: installed_apps, force=True)

        if hasattr(settings, 'RAVEN_CONFIG'):
            # Celery signal registration

            from raven import Client as RavenClient
            from raven.contrib.celery import register_signal as raven_register_signal
            from raven.contrib.celery import register_logger_signal as raven_register_logger_signal


            raven_client = RavenClient(dsn=settings.RAVEN_CONFIG['DSN'])
            raven_register_logger_signal(raven_client)
            raven_register_signal(raven_client)

def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--benchmark", help="the benchmark to run",
        type=benchmark_arg, default="latency",
    )
    parser.add_argument(
        "--count", help="the number of messages to benchmark with",
        type=int, default=10000,
    )
    parser.add_argument(
        "--use-green-threads", help="run workers with green threads rather than system threads",
        action="store_true", default=False,
    )
    parser.add_argument(
        "--use-celery", help="run the benchmark under Celery",
        action="store_true", default=False,
    )
    return parser.parse_args()