"""Notification emission helpers. Provides async (for routers) and sync (for Celery tasks) entry points to create notification rows in the audit_log table. """ import logging import uuid from datetime import datetime from sqlalchemy import create_engine, select from sqlalchemy.orm import Session from sqlalchemy.ext.asyncio import AsyncSession from app.domains.notifications.models import AuditLog logger = logging.getLogger(__name__) _engine = None def _get_engine(): global _engine if _engine is None: from app.config import settings as app_settings _engine = create_engine(app_settings.database_url_sync) return _engine async def emit_notification( db: AsyncSession, *, actor_user_id: str | uuid.UUID | None = None, target_user_id: str | uuid.UUID | None = None, action: str, entity_type: str | None = None, entity_id: str | None = None, details: dict | None = None, ) -> None: """Create a notification (async — for use inside FastAPI routers).""" try: entry = AuditLog( user_id=str(actor_user_id) if actor_user_id else None, target_user_id=str(target_user_id) if target_user_id else None, action=action, entity_type=entity_type, entity_id=str(entity_id) if entity_id else None, details=details, notification=True, timestamp=datetime.utcnow(), ) db.add(entry) await db.commit() except Exception: logger.exception("Failed to emit notification (async)") await db.rollback() def emit_notification_sync( *, actor_user_id: str | uuid.UUID | None = None, target_user_id: str | uuid.UUID | None = None, action: str, entity_type: str | None = None, entity_id: str | None = None, details: dict | None = None, ) -> None: """Create a notification (sync — for use inside Celery tasks).""" engine = _get_engine() try: with Session(engine) as session: entry = AuditLog( user_id=str(actor_user_id) if actor_user_id else None, target_user_id=str(target_user_id) if target_user_id else None, action=action, entity_type=entity_type, entity_id=str(entity_id) if entity_id else None, details=details, notification=True, timestamp=datetime.utcnow(), ) session.add(entry) session.commit() except Exception: logger.exception("Failed to emit notification (sync)") # ── Notification config helpers ───────────────────────────────────────────── def _is_channel_enabled_sync(user_id: str | None, event_type: str, channel: str) -> bool: """Check if a notification channel is enabled for a user (sync, for Celery).""" if not user_id: return channel == "in_app" # default: in_app on, email off engine = _get_engine() from app.domains.notifications.models import NotificationConfig with Session(engine) as session: cfg = session.execute( select(NotificationConfig).where( NotificationConfig.user_id == user_id, NotificationConfig.event_type == event_type, NotificationConfig.channel == channel, ) ).scalar_one_or_none() if cfg is None: return channel == "in_app" # default return cfg.enabled def send_email_notification_stub( *, to_user_id: str | None, event_type: str, subject: str, body: str, ) -> None: """Email notification stub — logs only, email sending not yet active.""" logger.info( "[EMAIL STUB] Would send email to user=%s event=%s subject=%s", to_user_id, event_type, subject ) async def get_notification_configs(db: AsyncSession, user_id: uuid.UUID) -> list: from app.domains.notifications.models import NotificationConfig from sqlalchemy import select as sa_select result = await db.execute( sa_select(NotificationConfig).where(NotificationConfig.user_id == user_id) .order_by(NotificationConfig.event_type, NotificationConfig.channel) ) return list(result.scalars().all()) async def upsert_notification_config( db: AsyncSession, user_id: uuid.UUID, event_type: str, channel: str, enabled: bool, ) -> object: from app.domains.notifications.models import NotificationConfig from sqlalchemy import select as sa_select result = await db.execute( sa_select(NotificationConfig).where( NotificationConfig.user_id == user_id, NotificationConfig.event_type == event_type, NotificationConfig.channel == channel, ) ) cfg = result.scalar_one_or_none() if cfg is None: cfg = NotificationConfig(user_id=user_id, event_type=event_type, channel=channel, enabled=enabled) db.add(cfg) else: cfg.enabled = enabled await db.commit() await db.refresh(cfg) return cfg async def reset_notification_configs(db: AsyncSession, user_id: uuid.UUID) -> list: from app.domains.notifications.models import NotificationConfig, NotificationEvent from sqlalchemy import delete as sa_delete await db.execute(sa_delete(NotificationConfig).where(NotificationConfig.user_id == user_id)) configs = [] for event in NotificationEvent.ALL: for channel, default_enabled in [("in_app", True), ("email", False)]: cfg = NotificationConfig( user_id=user_id, event_type=event, channel=channel, enabled=default_enabled ) db.add(cfg) configs.append(cfg) await db.commit() return configs