celery.events

celery.events

Events is a stream of messages sent for certain actions occurring in the worker (and clients if CELERY_SEND_TASK_SENT_EVENT is enabled), used for monitoring purposes.

class celery.events.Events(app=None)[源代码]

celery.events.Event(type, _fields=None, __dict__=<type ‘dict’>, __now__=<built-in function time>, \*fields*)[源代码]

Create an event.

An event is a dictionary, the only required field is type. A timestamp field will be set to the current time if not provided.

class celery.events.EventDispatcher(connection=None, hostname=None, enabled=True, channel=None, buffer_while_offline=True, app=None, serializer=None, groups=None)[源代码]

Dispatches event messages.

参数:
  • connection – Connection to the broker.
  • hostname – Hostname to identify ourselves as, by default uses the hostname returned by anon_nodename().
  • groups – List of groups to send events for. send() will ignore send requests to groups not in this list. If this is None, all events will be sent. Example groups include “task” and “worker”.
  • enabled – Set to False to not actually publish any events, making send() a noop operation.
  • channel – Can be used instead of connection to specify an exact channel to use when sending events.
  • buffer_while_offline – If enabled events will be buffered while the connection is down. flush() must be called as soon as the connection is re-established.

You need to close() this after use.

  • DISABLED_TRANSPORTS = set([‘sql’])

  • app = None

  • close()[源代码]

    Close the event dispatcher.

  • disable()[源代码]

  • enable()[源代码]

  • extend_buffer(other)

    Copies the outbound buffer of another instance.

  • flush()[源代码]

    Flushes the outbound buffer.

  • on_disabled = None

  • on_enabled = None

  • publish(type, fields, producer, retry=False, retry_policy=None, blind=False, utcoffset=<function utcoffset at 0xb126374>, Event=<function Event at 0xc2348b4>)[源代码]

    Publish event using a custom Producer instance.

    参数:
    • type – Event type name, with group separated by dash (-).
    • fields – Dictionary of event fields, must be json serializable.
    • producer – Producer instance to use, only the publish method will be called.
    • retry – Retry in the event of connection failure.
    • retry_policy – Dict of custom retry policy, see ensure().
    • blind – Don’t set logical clock value (also do not forward the internal logical clock).
    • Event – Event type used to create event, defaults to Event().
    • utcoffset – Function returning the current utcoffset in hours.

  • publisher None

  • send(type, blind=False, \*fields*)[源代码]

    Send event.

    参数:
    • type – Event type name, with group separated by dash (-).
    • retry – Retry in the event of connection failure.
    • retry_policy – Dict of custom retry policy, see ensure().
    • blind – Don’t set logical clock value (also do not forward the internal logical clock).
    • Event – Event type used to create event, defaults to Event().
    • utcoffset – Function returning the current utcoffset in hours.
    • **fields – Event fields, must be json serializable.

  • warn_if_yajl()

class celery.events.EventReceiver(channel, handlers=None, routing_key=’#’, node_id=None, app=None, queue_prefix=’celeryev’, accept=None)[源代码]

Capture events.

参数:
  • connection – Connection to the broker.
  • handlers – Event handlers.

handlers is a dict of event types and their handlers, the special handler “*” captures all events that doesn’t have a handler.

  • app = None

  • capture(limit=None, timeout=None, wakeup=True)[源代码]

    Open up a consumer capturing events.

    This has to run in the main process, and it will never stop unless forced via KeyboardInterrupt or SystemExit.

  • connection None

  • event_from_message(body, localize=True, now=<built-in function time>, tzfields=<operator.itemgetter object at 0xb9a19cc>, adjust_timestamp=<function adjust_timestamp at 0xb1263ac>, CLIENT_CLOCK_SKEW=-1)

  • get_consumers(Consumer, channel)

  • itercapture(limit=None, timeout=None, wakeup=True)[源代码]

  • on_consume_ready(connection, channel, consumers, wakeup=True, \*kwargs*)

  • process(type, event)[源代码]

    Process the received event by dispatching it to the appropriate handler.

  • wakeup_workers(channel=None)[源代码]