code: def snapshot_state(dag, execution_dates):
      TI = models.TaskInstance
      DR = models.DagRun
      with create_session() as session:
          return (
              session.query(TI)
              .join(TI.dag_run)
              .options(eagerload(TI.dag_run))
              .filter(
                  TI.dag_id == dag.dag_id,
                  DR.execution_date.in_(execution_dates),
              )
              .all()
          )
  
imports: from __future__ import annotations
  import datetime
  from typing import Callable
  import pytest
  from sqlalchemy.orm import eagerload
  from airflow import models
  from airflow.api.common.mark_tasks import _create_dagruns, _DagRunInfo, set_dag_run_state_to_failed, set_dag_run_state_to_queued, set_dag_run_state_to_running, set_dag_run_state_to_success, set_state
  from airflow.models import DagRun
  from airflow.utils import timezone
  from airflow.utils.session import create_session, provide_session
  from airflow.utils.state import State
  from airflow.utils.types import DagRunType
  from tests.test_utils.db import clear_db_runs
  from tests.test_utils.mapping import expand_mapped_task
label_desc: Use image node_id 3 for calling the TestMarkTasks obj's underlying member method code with example usage: obj.snapshot_state(dag, execution_dates) and returns: session