mystic blaze
I am trying to have signals using SQL Alchemy to use for WebSockets with FastAPI. I have now dumbed it down to try and just log a message whenever a new record is inserted but that isn't even working. Look below for relavent code.
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from app.core.config import settings
DATABASE_URL = settings.SQLALCHEMY_DATABASE_URI
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = async_sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
)
Base = declarative_base()
import logging
from sqlalchemy import event
from app.models import SMS
logger = logging.getLogger(__name__)
logger.info("Registering after_insert event listener for SMS.")
@event.listens_for(SMS, "after_insert")
def after_insert(mapper, connection, target):
logger.info(f"New SMS inserted: {target.id}")
iron juniper
compact snow
