code: def get_dataset_events(
      *,
      limit: int,
      offset: int = 0,
      order_by: str = "timestamp",
      dataset_id: int | None = None,
      source_dag_id: str | None = None,
      source_task_id: str | None = None,
      source_run_id: str | None = None,
      source_map_index: int | None = None,
      session: Session = NEW_SESSION,
  ) -> APIResponse:
      """Get dataset events."""
      allowed_attrs = [
          "source_dag_id",
          "source_task_id",
          "source_run_id",
          "source_map_index",
          "timestamp",
      ]
  
      query = select(DatasetEvent)
  
      if dataset_id:
          query = query.where(DatasetEvent.dataset_id == dataset_id)
      if source_dag_id:
          query = query.where(
              DatasetEvent.source_dag_id == source_dag_id
          )
      if source_task_id:
          query = query.where(
              DatasetEvent.source_task_id == source_task_id
          )
      if source_run_id:
          query = query.where(
              DatasetEvent.source_run_id == source_run_id
          )
      if source_map_index:
          query = query.where(
              DatasetEvent.source_map_index == source_map_index
          )
  
      query = query.options(subqueryload(DatasetEvent.created_dagruns))
  
      total_entries = get_query_count(query, session=session)
      query = apply_sorting(query, order_by, {}, allowed_attrs)
      events = session.scalars(query.offset(offset).limit(limit)).all()
      return dataset_event_collection_schema.dump(
          DatasetEventCollection(
              dataset_events=events, total_entries=total_entries
          )
      )
  
file_path: airflow/airflow/api_connexion/endpoints/dataset_endpoint.py
imports: from __future__ import annotations
  from typing import TYPE_CHECKING
  from sqlalchemy import func, select
  from sqlalchemy.orm import joinedload, subqueryload
  from airflow.api_connexion import security
  from airflow.api_connexion.exceptions import NotFound
  from airflow.api_connexion.parameters import apply_sorting, check_limit, format_parameters
  from airflow.api_connexion.schemas.dataset_schema import DatasetCollection, DatasetEventCollection, dataset_collection_schema, dataset_event_collection_schema, dataset_schema
  from airflow.models.dataset import DatasetEvent, DatasetModel
  from airflow.utils.db import get_query_count
  from airflow.utils.session import NEW_SESSION, provide_session
label_desc: Use image node_id 3 for calling a global function with example usage: get_dataset_events() and returns: dataset_event_collection_schema