class_docstr: the class to get Pool info.
class_docstr_tok: ["the","class","to","get","Pool","info","."]
docstr: Get Pool stats (Number of Running, Queued, Open & Total tasks).
  
  If ``lock_rows`` is True, and the database engine in use supports the ``NOWAIT`` syntax, then a
  non-blocking lock will be attempted -- if the lock is not available then SQLAlchemy will throw an
  OperationalError.
  
  :param lock_rows: Should we attempt to obtain a row-level lock on all the Pool rows returns
  :param session: SQLAlchemy ORM Session
docstr_tok: ["Get","Pool","stats","(","Number","of","Running",",","Queued",",","Open","&","Total","tasks",")",".","If","``","lock_rows","``","is","True",",","and","the","database","engine","in","use","supports","the","``","NOWAIT","``","syntax",",","then","a","non-blocking","lock","will","be","attempted","--","if","the","lock","is","not","available","then","SQLAlchemy","will","throw","an","OperationalError",".",":","param","lock_rows",":","Should","we","attempt","to","obtain","a","row-level","lock","on","all","the","Pool","rows","returns",":","param","session",":","SQLAlchemy","ORM","Session"]
code: def slots_stats(
      *,
      lock_rows: bool = False,
      session: Session = NEW_SESSION,
  ) -> dict[str, PoolStats]:
      """
      Get Pool stats (Number of Running, Queued, Open & Total tasks).
  
      If ``lock_rows`` is True, and the database engine in use supports the ``NOWAIT`` syntax, then a
      non-blocking lock will be attempted -- if the lock is not available then SQLAlchemy will throw an
      OperationalError.
  
      :param lock_rows: Should we attempt to obtain a row-level lock on all the Pool rows returns
      :param session: SQLAlchemy ORM Session
      """
      from airflow.models.taskinstance import (
          TaskInstance,
      )  # Avoid circular import
  
      pools: dict[str, PoolStats] = {}
      pool_includes_deferred: dict[str, bool] = {}
  
      query = select(Pool.pool, Pool.slots, Pool.include_deferred)
  
      if lock_rows:
          query = with_row_locks(
              query, session=session, **nowait(session)
          )
  
      pool_rows = session.execute(query)
      for pool_name, total_slots, include_deferred in pool_rows:
          if total_slots == -1:
              total_slots = float("inf")  # type: ignore
          pools[pool_name] = PoolStats(
              total=total_slots, running=0, queued=0, open=0, deferred=0
          )
          pool_includes_deferred[pool_name] = include_deferred
  
      allowed_execution_states = EXECUTION_STATES | {
          TaskInstanceState.DEFERRED,
      }
      state_count_by_pool = session.execute(
          select(
              TaskInstance.pool,
              TaskInstance.state,
              func.sum(TaskInstance.pool_slots),
          )
          .filter(TaskInstance.state.in_(allowed_execution_states))
          .group_by(TaskInstance.pool, TaskInstance.state)
      )
  
      # calculate queued and running metrics
      for pool_name, state, count in state_count_by_pool:
          # Some databases return decimal.Decimal here.
          count = int(count)
  
          stats_dict: PoolStats | None = pools.get(pool_name)
          if not stats_dict:
              continue
          # TypedDict key must be a string literal, so we use if-statements to set value
          if state == TaskInstanceState.RUNNING:
              stats_dict["running"] = count
          elif state == TaskInstanceState.QUEUED:
              stats_dict["queued"] = count
          elif state == TaskInstanceState.DEFERRED:
              stats_dict["deferred"] = count
          else:
              raise AirflowException(
                  f"Unexpected state. Expected values: {allowed_execution_states}."
              )
  
      # calculate open metric
      for pool_name, stats_dict in pools.items():
          stats_dict["open"] = (
              stats_dict["total"]
              - stats_dict["running"]
              - stats_dict["queued"]
          )
          if pool_includes_deferred[pool_name]:
              stats_dict["open"] -= stats_dict["deferred"]
  
      return pools
  
imports: from __future__ import annotations
  from typing import TYPE_CHECKING, Any
  from sqlalchemy import Boolean, Column, Integer, String, Text, func, select
  from airflow.exceptions import AirflowException, PoolNotFound
  from airflow.models.base import Base
  from airflow.ti_deps.dependencies_states import EXECUTION_STATES
  from airflow.typing_compat import TypedDict
  from airflow.utils.db import exists_query
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.sqlalchemy import nowait, with_row_locks
  from airflow.utils.state import TaskInstanceState
label_desc: Use image node_id 8 for calling the Pool obj's underlying member method code with example usage: obj.slots_stats() and returns: pools