Events - Feather

Events let one part of the app announce that something happened without knowing, or caring, who reacts to it. Creating a user is the usual example: the signup code shouldn’t need to know about your analytics.

Defining an event

from feather.events import Event

class UserCreatedEvent(Event):
    def __init__(self, user_id: str, email: str):
        super().__init__(user_id=user_id)
        self.email = email

Listening

from feather.events import listen

# synchronous, runs in the request thread
@listen(UserCreatedEvent)
def send_welcome_email(event):
    send_email(event.email, 'Welcome!')

# async, runs in a background thread pool
@listen(UserCreatedEvent, async_=True)
def track_signup_analytics(event):
    analytics.track('signup', user_id=event.user_id)

Dispatching

from feather.events import dispatch

@transactional
def create_user(self, email: str):
    user = User(email=email)
    self.db.add(user)
    dispatch(UserCreatedEvent(user_id=user.id, email=user.email))
    return user

Async listeners run in a ThreadPoolExecutor with four workers. They suit non-critical work such as analytics, logging and notifications. For anything that must not be lost, enqueue a background job.