"""
        from airflow.jobs.base_job import BaseJob  
        count = session.query(func.count(cls.id)).filter(cls.triggerer_id == triggerer_id).scalar()
        capacity -= count
        if capacity <= 0:
            return
        alive_triggerer_ids = [
            row[0]
            for row in session.query(BaseJob.id).filter(
                BaseJob.end_date.is_(None),
                BaseJob.latest_heartbeat > timezone.utcnow() - datetime.timedelta(seconds=30),
                BaseJob.job_type == "TriggererJob",
            )
        ]
        
        
        trigger_ids_query = with_row_locks(
            session.query(cls.id)
            .filter(or_(cls.triggerer_id.is_(None), cls.triggerer_id.notin_(alive_triggerer_ids)))
            .limit(capacity),
            session,
            skip_locked=True,
        ).all()
        if trigger_ids_query:
            session.query(cls).filter(cls.id.in_([i.id for i in trigger_ids_query])).update(
                {cls.triggerer_id: triggerer_id},
                synchronize_session=False,
            )
        session.commit()