class_docstr: Return Trigger Event if the inactivity period has passed with no increase in the number of objects.
  
  :param bucket: The Google Cloud Storage bucket where the objects are expected.
  :param prefix: The name of the prefix to check in the Google cloud storage bucket.
  :param poke_interval: polling period in seconds to check
  :param inactivity_period: The total seconds of inactivity to designate
      an upload session is over. Note, this mechanism is not real time and
      this operator may not return until a interval after this period
      has passed with no additional objects sensed.
  :param min_objects: The minimum number of objects needed for upload session
      to be considered valid.
  :param previous_objects: The set of object ids found during the last poke.
  :param allow_delete: Should this sensor consider objects being deleted
      between intervals valid behavior. If true a warning message will be logged
      when this happens. If false an error will be raised.
  :param google_cloud_conn_id: The connection ID to use when connecting
      to Google Cloud Storage.
code: def __init__(
      self,
      bucket: str,
      prefix: str,
      poke_interval: float,
      google_cloud_conn_id: str,
      hook_params: dict[str, Any],
      inactivity_period: float = 60 * 60,
      min_objects: int = 1,
      previous_objects: set[str] | None = None,
      allow_delete: bool = True,
  ):
      super().__init__(
          bucket=bucket,
          prefix=prefix,
          poke_interval=poke_interval,
          google_cloud_conn_id=google_cloud_conn_id,
          hook_params=hook_params,
      )
      self.inactivity_period = inactivity_period
      self.min_objects = min_objects
      self.previous_objects = previous_objects or set()
      self.inactivity_seconds = 0.0
      self.allow_delete = allow_delete
      self.last_activity_time: datetime | None = None
  
imports: from __future__ import annotations
  import asyncio
  import os
  from datetime import datetime
  from typing import Any, AsyncIterator
  from aiohttp import ClientSession
  from airflow.providers.google.cloud.hooks.gcs import GCSAsyncHook
  from airflow.triggers.base import BaseTrigger, TriggerEvent
  from airflow.utils import timezone
label_desc: Use image node_id 1 to create a new GCSUploadSessionTrigger object from inherited base classes: GCSPrefixBlobTrigger with example: obj = GCSUploadSessionTrigger(bucket, prefix, poke_interval, google_cloud_conn_id, hook_params, inactivity_period, min_objects, previous_objects, allow_delete)