Events

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

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)

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