In SQLAlchemy, events and event listeners provide a powerful mechanism for executing custom code in response to specific actions or changes within the ORM. These can be particularly useful for scenarios like tracking changes, sending notifications, or enforcing business logic. This blog will delve into the concept of events and event listeners in SQLAlchemy ORM using a practical example.

Understanding the Code Snippet

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

from sqlalchemy import Enum as EnumType

class Status(str, Enum):
    PENDING = "pending"
    REJECTED = "rejected"
    APPROVED = "approved"

class FuelRequest(AbstractBaseModel):
    __tablename__ = "fuels"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    public_id = Column(String(50), unique=True)
    status = Column(EnumType(Status), default=Status.PENDING)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", back_populates="fuels")

from sqlalchemy import event
from sqlalchemy.orm.attributes import get_history
from ..utils import get_db

def track_status_changes(mapper, connection, target):
    # Get the history of the 'status' attribute
    hist = get_history(target, 'status')

    if hist.has_changes():
        # Get the user's phone number
        db = next(get_db())
        user_phone_number = get_user_phone_number(db, user_id=target.user_id)

        # Create a message
        message = f"The status of your {target.__class__.__name__} Request has changed to {target.status}."
        # send_notification(user_phone_number, message)
        send_sms_notification(user_phone_number, message)

# Add the event listeners
event.listen(FuelRequest, 'after_update', track_status_changes)

In this example, an event listener named track_status_changes is defined. This listener is triggered after an update operation ('after_update') on instances of the FuelRequest classes. The listener checks if the ‘status’ attribute of the target object has changed. If so, it retrieves the user’s phone number and sends an SMS notification about the status change.

Breaking Down the Code

Defining the Event Listener

def track_status_changes(mapper, connection, target):
    # Get the history of the 'status' attribute
    hist = get_history(target, 'status')

    if hist.has_changes():
        # Get the user's phone number
        db = next(get_db())
        user_phone_number = get_user_phone_number(db, user_id=target.user_id)

        # Create a message to be sent
        message = f"The status of your {target.__class__.__name__} Request has changed to {target.status}."
        # Call the  send_sms_notification to send the message to the user using the users phone number
        send_sms_notification(user_phone_number, message)

The track_status_changes function is the event listener. It takes three

parameters: mapper, connection, and target.

The mapper parameter refers to ORM Mapper for the object.

The connection parameter refers to the database connection in use.

The target parameter refers to the instance of the class (e.g. FuelRequest) that triggered the event.

Inside the listener, it checks if the ‘status’ attribute has changed and, if so, retrieves the user’s phone number and sends an SMS notification.

import os
from twilio.rest import Client
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

account_sid = os.environ['ACCOUNT_SID']
auth_token = os.environ['AUTH_TOKEN']
phone_number = os.environ['TWILIO_PHONE_NUMBER']
client = Client(account_sid, auth_token)

def get_user_phone_number(db, user_id: int):
    # get the user
    user = db.query(User).filter(User.id == user_id).first()
    # return the user's phone number
    return user.phone_number if user else None

def send_sms_notification(user_phone_number, message_body):
    try:
        message = client.messages.create(
            body=message_body,
            from_= phone_number,  # Your Twilio phone number
            to=user_phone_number  # Get the user phone numbr from the User model
        )
    except Exception as e:
        raise ValueError("Error occurred while sending message: ", e)

Adding Event Listeners

event.listen(FuelRequest, 'after_update', track_status_changes)

The event.listen method is used to attach the track_status_changes listener to the

‘after_update’ event of the specified classes ( FuelRequest). This means that the listener will be

executed every time an instance of these classes undergoes an update operation.

Conclusion

Events and event listeners in SQLAlchemy provide a flexible way to incorporate custom behavior into your ORM-based applications. Whether you need to track changes, enforce business rules, or send notifications, event listeners offer a clean and modular solution. The provided code snippet demonstrates how to use event listeners to respond to updates in the ‘status’ attribute of different ORM entities. Feel free to adapt and extend this example to suit your specific application needs. Happy coding! 🚀💻