Шаблон наблюдателя Python: примеры, советы?

Ответ 1

Однако он не обладает гибкостью.

Ну... на самом деле, это похоже на хороший дизайн для меня, если асинхронный API - это то, что вы хотите. Обычно это так. Возможно, вам нужно перейти от stderr к Python logging, который имеет своего рода модель публикации/подписки, что с Logger.addHandler() и т.д.

Если вы хотите поддержать наблюдателей, я советую сохранить это просто. Вам действительно нужны только несколько строк кода.

class Event(object):
    pass

class Observable(object):
    def __init__(self):
        self.callbacks = []
    def subscribe(self, callback):
        self.callbacks.append(callback)
    def fire(self, **attrs):
        e = Event()
        e.source = self
        for k, v in attrs.iteritems():
            setattr(e, k, v)
        for fn in self.callbacks:
            fn(e)

Ваш класс Job может быть подклассом Observable. Когда что-то интересное, вызовите self.fire(type="progress", percent=50) или тому подобное.

Ответ 2

Еще несколько подходов...

Пример: модуль регистрации

Возможно, вам нужно перейти от stderr к Python logging, который имеет мощную модель публикации/подписки.

Легко начать создавать записи журнала.

# producer
import logging

log = logging.getLogger("myjobs")  # that all the setup you need

class MyJob(object):
    def run(self):
        log.info("starting job")
        n = 10
        for i in range(n):
            log.info("%.1f%% done" % (100.0 * i / n))
        log.info("work complete")

На стороне потребителя есть немного больше работы. К сожалению, для конфигурирования вывода регистратора требуется, например, 7 целых строк кода.;)

# consumer
import myjobs, sys, logging

if user_wants_log_output:
    ch = logging.StreamHandler(sys.stderr)
    ch.setLevel(logging.INFO)
    formatter = logging.Formatter(
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s")
    ch.setFormatter(formatter)
    myjobs.log.addHandler(ch)
    myjobs.log.setLevel(logging.INFO)

myjobs.MyJob().run()

С другой стороны, в пакете регистрации есть удивительное количество вещей. Если вам когда-либо понадобится отправить данные журнала во вращающийся набор файлов, адрес электронной почты и журнал событий Windows, вы будете защищены.

Пример: простейший возможный наблюдатель

Но вам вообще не нужно использовать какую-либо библиотеку. Очень простой способ поддержать наблюдателей - вызвать метод, который ничего не делает.

# producer
class MyJob(object):
    def on_progress(self, pct):
        """Called when progress is made. pct is the percent complete.
        By default this does nothing. The user may override this method
        or even just assign to it."""
        pass

    def run(self):
        n = 10
        for i in range(n):
            self.on_progress(100.0 * i / n)
        self.on_progress(100.0)

# consumer
import sys, myjobs
job = myjobs.MyJob()
job.on_progress = lambda pct: sys.stdout.write("%.1f%% done\n" % pct)
job.run()

Иногда вместо написания лямбда вы можете просто сказать job.on_progress = progressBar.update, что приятно.

Это примерно так же просто, как и получается. Один из недостатков заключается в том, что он, естественно, не поддерживает нескольких слушателей, которые подписываются на одни и те же события.

Пример: С# -подобные события

С небольшим количеством кода поддержки вы можете получить С# -подобные события в Python. Здесь код:

# glue code
class event(object):
    def __init__(self, func):
        self.__doc__ = func.__doc__
        self._key = ' ' + func.__name__
    def __get__(self, obj, cls):
        try:
            return obj.__dict__[self._key]
        except KeyError, exc:
            be = obj.__dict__[self._key] = boundevent()
            return be

class boundevent(object):
    def __init__(self):
        self._fns = []
    def __iadd__(self, fn):
        self._fns.append(fn)
        return self
    def __isub__(self, fn):
        self._fns.remove(fn)
        return self
    def __call__(self, *args, **kwargs):
        for f in self._fns[:]:
            f(*args, **kwargs)

Производитель объявляет событие с помощью декоратора:

# producer
class MyJob(object):
    @event
    def progress(pct):
        """Called when progress is made. pct is the percent complete."""

    def run(self):
        n = 10
        for i in range(n+1):
            self.progress(100.0 * i / n)

#consumer
import sys, myjobs
job = myjobs.MyJob()
job.progress += lambda pct: sys.stdout.write("%.1f%% done\n" % pct)
job.run()

Это работает точно так же, как и код "простого наблюдателя" выше, но вы можете добавить столько слушателей, сколько хотите, используя +=. (В отличие от С#, нет типов обработчиков событий, вам не нужно new EventHandler(foo.bar) при подписке на событие, и вам не нужно проверять значение null перед запуском события. Подобно С#, события не прерывают исключения. )

Как выбрать

Если logging делает все, что вам нужно, используйте это. В противном случае сделайте самое простое, что работает для вас. Главное отметить, что вам не нужно принимать большую внешнюю зависимость.

Ответ 3

Я думаю, что люди в других ответах переусердствовали. Вы можете легко достичь событий в Python с менее чем 15 строками кода.

У вас просто два класса: Event и Observer. Любой класс, который хочет прослушать событие, должен унаследовать Observer и настроить прослушивание (наблюдение) для определенного события. Когда Event создается и запускается, все наблюдатели, слушающие это событие, будут выполнять указанные функции обратного вызова.

class Observer():
    _observers = []
    def __init__(self):
        self._observers.append(self)
        self._observables = {}
    def observe(self, event_name, callback):
        self._observables[event_name] = callback


class Event():
    def __init__(self, name, data, autofire = True):
        self.name = name
        self.data = data
        if autofire:
            self.fire()
    def fire(self):
        for observer in Observer._observers:
            if self.name in observer._observables:
                observer._observables[self.name](self.data)

Пример:

class Room(Observer):

    def __init__(self):
        print("Room is ready.")
        Observer.__init__(self) # Observer init needs to be called
    def someone_arrived(self, who):
        print(who + " has arrived!")

room = Room()
room.observe('someone arrived',  room.someone_arrived)

Event('someone arrived', 'Lenard')

Вывод:

Room is ready.
Lenard has arrived!

Ответ 4

От wikipedia:

from collections import defaultdict

class Observable (defaultdict):

  def __init__ (self):
      defaultdict.__init__(self, object)

  def emit (self, *args):
      '''Pass parameters to all observers and update states.'''
      for subscriber in self:
          response = subscriber(*args)
          self[subscriber] = response

  def subscribe (self, subscriber):
      '''Add a new subscriber to self.'''
      self[subscriber]

  def stat (self):
      '''Return a tuple containing the state of each observer.'''
      return tuple(self.values())

Наблюдаемый используется так.

myObservable = Observable ()

# subscribe some inlined functions.
# myObservable[lambda x, y: x * y] would also work here.
myObservable.subscribe(lambda x, y: x * y)
myObservable.subscribe(lambda x, y: float(x) / y)
myObservable.subscribe(lambda x, y: x + y)
myObservable.subscribe(lambda x, y: x - y)

# emit parameters to each observer
myObservable.emit(6, 2)

# get updated values
myObservable.stat()         # returns: (8, 3.0, 4, 12)

Ответ 5

Как насчет реализации, где объекты не поддерживаются только потому, что они что-то наблюдают? Ниже приведена реализация шаблона наблюдателя со следующими функциями:

• Использование pythonic. Чтобы добавить наблюдателя к связанному методу .bar экземпляра foo, просто выполните foo.bar.addObserver(observer).
• Наблюдатели не остаются в живых в силу того, что они наблюдатели. Другими словами, код наблюдателя не использует сильные ссылки.
• Не требуется никакого подкласса (дескрипторы ftw).
• Может использоваться с нераспадаемыми типами.
• Может использоваться столько раз, сколько вы хотите в одном классе.
• (bonus) На сегодняшний день код существует в правильном загружаемом, устанавливаемом пакете на github.

Здесь код (пакет github или пакет PyPI имеют самую актуальную реализацию):

import weakref
import functools

class ObservableMethod(object):
    """
    A proxy for a bound method which can be observed.

    I behave like a bound method, but other bound methods can subscribe to be
    called whenever I am called.
    """

    def __init__(self, obj, func):
        self.func = func
        functools.update_wrapper(self, func)
        self.objectWeakRef = weakref.ref(obj)
        self.callbacks = {}  #observing object ID -> weak ref, methodNames

    def addObserver(self, boundMethod):
        """
        Register a bound method to observe this ObservableMethod.

        The observing method will be called whenever this ObservableMethod is
        called, and with the same arguments and keyword arguments. If a
        boundMethod has already been registered to as a callback, trying to add
        it again does nothing. In other words, there is no way to sign up an
        observer to be called back multiple times.
        """
        obj = boundMethod.__self__
        ID = id(obj)
        if ID in self.callbacks:
            s = self.callbacks[ID][1]
        else:
            wr = weakref.ref(obj, Cleanup(ID, self.callbacks))
            s = set()
            self.callbacks[ID] = (wr, s)
        s.add(boundMethod.__name__)

    def discardObserver(self, boundMethod):
        """
        Un-register a bound method.
        """
        obj = boundMethod.__self__
        if id(obj) in self.callbacks:
            self.callbacks[id(obj)][1].discard(boundMethod.__name__)

    def __call__(self, *arg, **kw):
        """
        Invoke the method which I proxy, and all of it callbacks.

        The callbacks are called with the same *args and **kw as the main
        method.
        """
        result = self.func(self.objectWeakRef(), *arg, **kw)
        for ID in self.callbacks:
            wr, methodNames = self.callbacks[ID]
            obj = wr()
            for methodName in methodNames:
                getattr(obj, methodName)(*arg, **kw)
        return result

    @property
    def __self__(self):
        """
        Get a strong reference to the object owning this ObservableMethod

        This is needed so that ObservableMethod instances can observe other
        ObservableMethod instances.
        """
        return self.objectWeakRef()


class ObservableMethodDescriptor(object):

    def __init__(self, func):
        """
        To each instance of the class using this descriptor, I associate an
        ObservableMethod.
        """
        self.instances = {}  # Instance id -> (weak ref, Observablemethod)
        self._func = func

    def __get__(self, inst, cls):
        if inst is None:
            return self
        ID = id(inst)
        if ID in self.instances:
            wr, om = self.instances[ID]
            if not wr():
                msg = "Object id %d should have been cleaned up"%(ID,)
                raise RuntimeError(msg)
        else:
            wr = weakref.ref(inst, Cleanup(ID, self.instances))
            om = ObservableMethod(inst, self._func)
            self.instances[ID] = (wr, om)
        return om

    def __set__(self, inst, val):
        raise RuntimeError("Assigning to ObservableMethod not supported")


def event(func):
    return ObservableMethodDescriptor(func)


class Cleanup(object):
    """
    I manage remove elements from a dict whenever I'm called.

    Use me as a weakref.ref callback to remove an object id from a dict
    when that object is garbage collected.
    """
    def __init__(self, key, d):
        self.key = key
        self.d = d

    def __call__(self, wr):
        del self.d[self.key]

Чтобы использовать это, мы просто украшаем методы, которые хотим сделать наблюдаемыми с помощью @event. Вот пример

class Foo(object):
    def __init__(self, name):
        self.name = name

    @event
    def bar(self):
        print("%s called bar"%(self.name,))

    def baz(self):
        print("%s called baz"%(self.name,))

a = Foo('a')
b = Foo('b')
a.bar.addObserver(b.bar)
a.bar()

Ответ 6

Основываясь на ответе Джейсона, я применил пример событий, подобных С#, как полноценный модуль python, включая документацию и тесты. Мне нравится фантастический питонический материал:)

Итак, если вам нужно какое-то готовое к использованию решение, вы можете просто использовать код в github.

Ответ 7

Пример: скрученные наблюдатели журналов

Чтобы зарегистрировать наблюдателя yourCallable() (вызываемый, который принимает словарь), чтобы получать все события журнала (в дополнение к любым другим наблюдателям):

twisted.python.log.addObserver(yourCallable)

Пример: полный пример производителя/потребителя

Из списка рассылки Twisted-Python:

#!/usr/bin/env python
"""Serve as a sample implementation of a twisted producer/consumer
system, with a simple TCP server which asks the user how many random
integers they want, and it sends the result set back to the user, one
result per line."""

import random

from zope.interface import implements
from twisted.internet import interfaces, reactor
from twisted.internet.protocol import Factory
from twisted.protocols.basic import LineReceiver

class Producer:
    """Send back the requested number of random integers to the client."""
    implements(interfaces.IPushProducer)
    def __init__(self, proto, cnt):
        self._proto = proto
        self._goal = cnt
        self._produced = 0
        self._paused = False
    def pauseProducing(self):
        """When we've produced data too fast, pauseProducing() will be
called (reentrantly from within resumeProducing transport.write
method, most likely), so set a flag that causes production to pause
temporarily."""
        self._paused = True
        print('pausing connection from %s' % (self._proto.transport.getPeer()))
    def resumeProducing(self):
        self._paused = False
        while not self._paused and self._produced < self._goal:
            next_int = random.randint(0, 10000)
            self._proto.transport.write('%d\r\n' % (next_int))
            self._produced += 1
        if self._produced == self._goal:
            self._proto.transport.unregisterProducer()
            self._proto.transport.loseConnection()
    def stopProducing(self):
        pass

class ServeRandom(LineReceiver):
    """Serve up random data."""
    def connectionMade(self):
        print('connection made from %s' % (self.transport.getPeer()))
        self.transport.write('how many random integers do you want?\r\n')
    def lineReceived(self, line):
        cnt = int(line.strip())
        producer = Producer(self, cnt)
        self.transport.registerProducer(producer, True)
        producer.resumeProducing()
    def connectionLost(self, reason):
        print('connection lost from %s' % (self.transport.getPeer()))
factory = Factory()
factory.protocol = ServeRandom
reactor.listenTCP(1234, factory)
print('listening on 1234...')
reactor.run()

Ответ 8

Функциональный подход к дизайну наблюдателя:

def add_listener(obj, method_name, listener):

    # Get any existing listeners
    listener_attr = method_name + '_listeners'
    listeners = getattr(obj, listener_attr, None)

    # If this is the first listener, then set up the method wrapper
    if not listeners:

        listeners = [listener]
        setattr(obj, listener_attr, listeners)

        # Get the object method
        method = getattr(obj, method_name)

        @wraps(method)
        def method_wrapper(*args, **kwags):
            method(*args, **kwags)
            for l in listeners:
                l(obj, *args, **kwags) # Listener also has object argument

        # Replace the original method with the wrapper
        setattr(obj, method_name, method_wrapper)

    else:
        # Event is already set up, so just add another listener
        listeners.append(listener)


def remove_listener(obj, method_name, listener):

    # Get any existing listeners
    listener_attr = method_name + '_listeners'
    listeners = getattr(obj, listener_attr, None)

    if listeners:
        # Remove the listener
        next((listeners.pop(i)
              for i, l in enumerate(listeners)
              if l == listener),
             None)

        # If this was the last listener, then remove the method wrapper
        if not listeners:
            method = getattr(obj, method_name)
            delattr(obj, listener_attr)
            setattr(obj, method_name, method.__wrapped__)

Эти методы затем могут использоваться для добавления слушателя к любому методу класса. Например:

class MyClass(object):

    def __init__(self, prop):
        self.prop = prop

    def some_method(self, num, string):
        print('method:', num, string)

def listener_method(obj, num, string):
    print('listener:', num, string, obj.prop)

my = MyClass('my_prop')

add_listener(my, 'some_method', listener_method)
my.some_method(42, 'with listener')

remove_listener(my, 'some_method', listener_method)
my.some_method(42, 'without listener')

И результат:

method: 42 with listener
listener: 42 with listener my_prop
method: 42 without listener