class_docstr: Dep for pool slots availability.
class_docstr_tok: ["Dep","for","pool","slots","availability","."]
docstr: Determine if the pool task instance is in has available slots.
  
  :param ti: the task instance to get the dependency status for
  :param session: database session
  :param dep_context: the context for which this dependency should be evaluated for
  :return: True if there are available slots in the pool.
docstr_tok: ["Determine","if","the","pool","task","instance","is","in","has","available","slots",".",":","param","ti",":","the","task","instance","to","get","the","dependency","status","for",":","param","session",":","database","session",":","param","dep_context",":","the","context","for","which","this","dependency","should","be","evaluated","for",":","return",":","True","if","there","are","available","slots","in","the","pool","."]
code: def _get_dep_statuses(self, ti, session, dep_context=None):
      """
      Determine if the pool task instance is in has available slots.
  
      :param ti: the task instance to get the dependency status for
      :param session: database session
      :param dep_context: the context for which this dependency should be evaluated for
      :return: True if there are available slots in the pool.
      """
      from airflow.models.pool import (
          Pool,
      )  # To avoid a circular dependency
  
      pool_name = ti.pool
  
      # Controlled by UNIQUE key in slot_pool table, only (at most) one result can be returned.
      pool: Pool | None = session.scalar(
          select(Pool).where(Pool.pool == pool_name)
      )
      if pool is None:
          yield self._failing_status(
              reason=f"Tasks using non-existent pool '{pool_name}' will not be scheduled"
          )
          return
  
      open_slots = pool.open_slots(session=session)
      if ti.state in pool.get_occupied_states():
          open_slots += ti.pool_slots
  
      if open_slots <= (ti.pool_slots - 1):
          yield self._failing_status(
              reason=f"Not scheduling since there are {open_slots} open slots in pool {pool_name} "
              f"and require {ti.pool_slots} pool slots"
          )
      else:
          yield self._passing_status(
              reason=f"There are enough open slots in {pool_name} to execute the task",
          )
  
imports: from __future__ import annotations
  from sqlalchemy import select
  from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
  from airflow.utils.session import provide_session
label_desc: Use image node_id 1 for calling the PoolSlotsAvailableDep obj's underlying member method code with example usage: obj._get_dep_statuses(ti, session, dep_context) without return types