class_docstr: Custom security manager, which introduces a permission model adapted to Airflow.
  
  It's named V2 to differentiate it from the obsolete airflow.www.security.AirflowSecurityManager.
class_docstr_tok: ["Custom","security","manager",",","which","introduces","a","permission","model","adapted","to","Airflow",".","It","'s","named","V2","to","differentiate","it","from","the","obsolete","airflow.www.security.AirflowSecurityManager","."]
docstr: Return the map associating a FAB resource name to the corresponding auth manager is_authorized_ API.
  
  The function returned takes the FAB action name and the user as parameter.
docstr_tok: ["Return","the","map","associating","a","FAB","resource","name","to","the","corresponding","auth","manager","is_authorized_","API",".","The","function","returned","takes","the","FAB","action","name","and","the","user","as","parameter","."]
code: def _auth_manager_is_authorized_map(
      self, session: Session = NEW_SESSION
  ) -> dict[str, Callable[[str, str | None, BaseUser | None], bool]]:
      """
      Return the map associating a FAB resource name to the corresponding auth manager is_authorized_ API.
  
      The function returned takes the FAB action name and the user as parameter.
      """
      auth_manager = get_auth_manager()
      methods = get_method_from_fab_action_map()
  
      def get_connection_id(resource_pk):
          if not resource_pk:
              return None
          connection = session.scalar(
              select(Connection)
              .where(Connection.id == resource_pk)
              .limit(1)
          )
          if not connection:
              raise AirflowException("Connection not found")
          return connection.conn_id
  
      def get_dag_id_from_dagrun_id(resource_pk):
          if not resource_pk:
              return None
          dagrun = session.scalar(
              select(DagRun).where(DagRun.id == resource_pk).limit(1)
          )
          if not dagrun:
              raise AirflowException("DagRun not found")
          return dagrun.dag_id
  
      def get_dag_id_from_task_instance(resource_pk):
          if not resource_pk:
              return None
          composite_pk = json.loads(resource_pk)
          ti = session.scalar(
              select(DagRun)
              .where(
                  TaskInstance.dag_id == composite_pk[0],
                  TaskInstance.task_id == composite_pk[1],
                  TaskInstance.run_id == composite_pk[2],
                  TaskInstance.map_index >= composite_pk[3],
              )
              .limit(1)
          )
          if not ti:
              raise AirflowException("Task instance not found")
          return ti.dag_id
  
      def get_pool_name(resource_pk):
          if not resource_pk:
              return None
          pool = session.scalar(
              select(Pool).where(Pool.id == resource_pk).limit(1)
          )
          if not pool:
              raise AirflowException("Pool not found")
          return pool.pool
  
      def get_variable_key(resource_pk):
          if not resource_pk:
              return None
          variable = session.scalar(
              select(Variable)
              .where(Variable.id == resource_pk)
              .limit(1)
          )
          if not variable:
              raise AirflowException("Connection not found")
          return variable.key
  
      return {
          RESOURCE_AUDIT_LOG: lambda action, resource_pk, user: auth_manager.is_authorized_dag(
              method=methods[action],
              access_entity=DagAccessEntity.AUDIT_LOG,
              user=user,
          ),
          RESOURCE_CLUSTER_ACTIVITY: lambda action, resource_pk, user: auth_manager.is_authorized_view(
              access_view=AccessView.CLUSTER_ACTIVITY,
              user=user,
          ),
          RESOURCE_CONFIG: lambda action, resource_pk, user: auth_manager.is_authorized_configuration(
              method=methods[action],
              user=user,
          ),
          RESOURCE_CONNECTION: lambda action, resource_pk, user: auth_manager.is_authorized_connection(
              method=methods[action],
              details=ConnectionDetails(
                  conn_id=get_connection_id(resource_pk)
              ),
              user=user,
          ),
          RESOURCE_DAG: lambda action, resource_pk, user: auth_manager.is_authorized_dag(
              method=methods[action],
              user=user,
          ),
          RESOURCE_DAG_CODE: lambda action, resource_pk, user: auth_manager.is_authorized_dag(
              method=methods[action],
              access_entity=DagAccessEntity.CODE,
              user=user,
          ),
          RESOURCE_DAG_DEPENDENCIES: lambda action, resource_pk, user: auth_manager.is_authorized_dag(
              method=methods[action],
              access_entity=DagAccessEntity.DEPENDENCIES,
              user=user,
          ),
          RESOURCE_DAG_RUN: lambda action, resource_pk, user: auth_manager.is_authorized_dag(
              method=methods[action],
              access_entity=DagAccessEntity.RUN,
              details=DagDetails(
                  id=get_dag_id_from_dagrun_id(resource_pk)
              ),
              user=user,
          ),
          RESOURCE_DATASET: lambda action, resource_pk, user: auth_manager.is_authorized_dataset(
              method=methods[action],
              user=user,
          ),
          RESOURCE_DOCS: lambda action, resource_pk, user: auth_manager.is_authorized_view(
              access_view=AccessView.DOCS,
              user=user,
          ),
          RESOURCE_PLUGIN: lambda action, resource_pk, user: auth_manager.is_authorized_view(
              access_view=AccessView.PLUGINS,
              user=user,
          ),
          RESOURCE_JOB: lambda action, resource_pk, user: auth_manager.is_authorized_view(
              access_view=AccessView.JOBS,
              user=user,
          ),
          RESOURCE_POOL: lambda action, resource_pk, user: auth_manager.is_authorized_pool(
              method=methods[action],
              details=PoolDetails(name=get_pool_name(resource_pk)),
              user=user,
          ),
          RESOURCE_PROVIDER: lambda action, resource_pk, user: auth_manager.is_authorized_view(
              access_view=AccessView.PROVIDERS,
              user=user,
          ),
          RESOURCE_SLA_MISS: lambda action, resource_pk, user: auth_manager.is_authorized_dag(
              method=methods[action],
              access_entity=DagAccessEntity.SLA_MISS,
              user=user,
          ),
          RESOURCE_TASK_INSTANCE: lambda action, resource_pk, user: auth_manager.is_authorized_dag(
              method=methods[action],
              access_entity=DagAccessEntity.TASK_INSTANCE,
              details=DagDetails(
                  id=get_dag_id_from_task_instance(resource_pk)
              ),
              user=user,
          ),
          RESOURCE_TASK_RESCHEDULE: lambda action, resource_pk, user: auth_manager.is_authorized_dag(
              method=methods[action],
              access_entity=DagAccessEntity.TASK_RESCHEDULE,
              user=user,
          ),
          RESOURCE_TRIGGER: lambda action, resource_pk, user: auth_manager.is_authorized_view(
              access_view=AccessView.TRIGGERS,
              user=user,
          ),
          RESOURCE_VARIABLE: lambda action, resource_pk, user: auth_manager.is_authorized_variable(
              method=methods[action],
              details=VariableDetails(
                  key=get_variable_key(resource_pk)
              ),
              user=user,
          ),
          RESOURCE_XCOM: lambda action, resource_pk, user: auth_manager.is_authorized_dag(
              method=methods[action],
              access_entity=DagAccessEntity.XCOM,
              user=user,
          ),
      }
  
imports: from __future__ import annotations
  import json
  from functools import cached_property
  from typing import TYPE_CHECKING, Callable
  from flask import g
  from flask_limiter import Limiter
  from flask_limiter.util import get_remote_address
  from sqlalchemy import select
  from airflow.auth.managers.models.resource_details import AccessView, ConnectionDetails, DagAccessEntity, DagDetails, PoolDetails, VariableDetails
  from airflow.auth.managers.utils.fab import get_method_from_fab_action_map
  from airflow.exceptions import AirflowException
  from airflow.models import Connection, DagRun, Pool, TaskInstance, Variable
  from airflow.security.permissions import ACTION_CAN_ACCESS_MENU, ACTION_CAN_READ, RESOURCE_ADMIN_MENU, RESOURCE_AUDIT_LOG, RESOURCE_BROWSE_MENU, RESOURCE_CLUSTER_ACTIVITY, RESOURCE_CONFIG, RESOURCE_CONNECTION, RESOURCE_DAG, RESOURCE_DAG_CODE, RESOURCE_DAG_DEPENDENCIES, RESOURCE_DAG_RUN, RESOURCE_DATASET, RESOURCE_DOCS, RESOURCE_DOCS_MENU, RESOURCE_JOB, RESOURCE_PLUGIN, RESOURCE_POOL, RESOURCE_PROVIDER, RESOURCE_SLA_MISS, RESOURCE_TASK_INSTANCE, RESOURCE_TASK_RESCHEDULE, RESOURCE_TRIGGER, RESOURCE_VARIABLE, RESOURCE_XCOM
  from airflow.utils.log.logging_mixin import LoggingMixin
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.www.extensions.init_auth_manager import get_auth_manager
  from airflow.www.utils import CustomSQLAInterface
label_desc: Use image node_id 8 for calling the AirflowSecurityManagerV2 obj's underlying member method code with example usage: obj._auth_manager_is_authorized_map(session) and returns: dict, connection, dagrun, ti, pool, variable, None, None, None, None, None