def filter_snooze()

in src/dispatch/signal/service.py [0:0]

• 31 lines of code
• 12 McCabe index (conditional complexity)

def filter_snooze(*, db_session: Session, signal_instance: SignalInstance) -> SignalInstance:
    """Filters a signal instance for snoozing.

    Args:
        db_session (Session): Database session.
        signal_instance (SignalInstance): Signal instance to be filtered.

    Returns:
        SignalInstance: The filtered signal instance.
    """
    for f in signal_instance.signal.filters:
        if not f.mode:
            log.warning(f"Signal filter {f.name} has no mode")
            continue

        if f.mode != SignalFilterMode.active:
            continue

        if not f.action:
            log.warning(f"Signal filter {f.name} has no action")
            continue

        if f.action != SignalFilterAction.snooze:
            continue

        if not f.expiration:
            log.warning(f"Signal filter {f.name} has no expiration date")
            continue

        if f.expiration.replace(tzinfo=timezone.utc) <= datetime.now(timezone.utc):
            continue

        query = db_session.query(SignalInstance).filter(
            SignalInstance.signal_id == signal_instance.signal_id
        )
        query = apply_filter_specific_joins(SignalInstance, f.expression, query)
        query = apply_filters(query, f.expression)
        # an expression is not required for snoozing, if absent we snooze regardless of entity
        if f.expression:
            instances = query.filter(SignalInstance.id == signal_instance.id).all()

            if instances:
                signal_instance.filter_action = SignalFilterAction.snooze
                break
        else:
            signal_instance.filter_action = SignalFilterAction.snooze
            break

    return signal_instance