class_docstr: This class is deprecated, please use :class:`airflow.utils.task_group.TaskGroup`.
  
  This runs a sub dag. By convention, a sub dag's dag_id
  should be prefixed by its parent and a dot. As in `parent.child`.
  Although SubDagOperator can occupy a pool/concurrency slot,
  user can specify the mode=reschedule so that the slot will be
  released periodically to avoid potential deadlock.
  
  :param subdag: the DAG object to run as a subdag of the current DAG.
  :param session: sqlalchemy session
  :param conf: Configuration for the subdag
  :param propagate_skipped_state: by setting this argument you can define
      whether the skipped state of leaf task(s) should be propagated to the
      parent dag's downstream task.
class_docstr_tok: ["This","class","is","deprecated",",","please","use",":","class",":","`","airflow.utils.task_group.TaskGroup","`",".","This","runs","a","sub","dag",".","By","convention",",","a","sub","dag","'s","dag_id","should","be","prefixed","by","its","parent","and","a","dot",".","As","in","`","parent.child","`",".","Although","SubDagOperator","can","occupy","a","pool\/concurrency","slot",",","user","can","specify","the","mode=reschedule","so","that","the","slot","will","be","released","periodically","to","avoid","potential","deadlock",".",":","param","subdag",":","the","DAG","object","to","run","as","a","subdag","of","the","current","DAG",".",":","param","session",":","sqlalchemy","session",":","param","conf",":","Configuration","for","the","subdag",":","param","propagate_skipped_state",":","by","setting","this","argument","you","can","define","whether","the","skipped","state","of","leaf","task","(","s",")","should","be","propagated","to","the","parent","dag","'s","downstream","task","."]
code: def _validate_pool(self, session):
      if self.pool:
          conflicts = [
              t for t in self.subdag.tasks if t.pool == self.pool
          ]
          if conflicts:
              # only query for pool conflicts if one may exist
              pool = session.scalar(
                  select(Pool).where(
                      Pool.slots == 1, Pool.pool == self.pool
                  )
              )
              if pool and any(
                  t.pool == self.pool for t in self.subdag.tasks
              ):
                  raise AirflowException(
                      f"SubDagOperator {self.task_id} and subdag task{'s' if len(conflicts) > 1 else ''} "
                      f"{', '.join(t.task_id for t in conflicts)} both use pool {self.pool}, "
                      f"but the pool only has 1 slot. The subdag tasks will never run."
                  )
  
imports: from __future__ import annotations
  import warnings
  from enum import Enum
  from typing import TYPE_CHECKING
  from sqlalchemy import select
  from airflow.api.common.experimental.get_task_instance import get_task_instance
  from airflow.exceptions import AirflowException, RemovedInAirflow3Warning, TaskInstanceNotFound
  from airflow.models import DagRun
  from airflow.models.dag import DagContext
  from airflow.models.pool import Pool
  from airflow.models.taskinstance import TaskInstance
  from airflow.sensors.base import BaseSensorOperator
  from airflow.utils.session import NEW_SESSION, create_session, provide_session
  from airflow.utils.state import DagRunState, TaskInstanceState
  from airflow.utils.types import DagRunType
label_desc: Use image node_id 3 for calling the SubDagOperator obj's underlying member method code with example usage: obj._validate_pool(session) without return types