code: def get_fs(
      conn_id: str | None, storage_options: dict[str, str] | None = None
  ) -> AbstractFileSystem:
      try:
          from s3fs import S3FileSystem
      except ImportError:
          raise ImportError(
              "Airflow FS S3 protocol requires the s3fs library, but it is not installed as it requires"
              "aiobotocore. Please install the s3 protocol support library by running: "
              "pip install apache-airflow-providers-amazon[s3fs]"
          )
  
      s3_hook = S3Hook(aws_conn_id=conn_id)
      session = s3_hook.get_session(deferrable=True)
      endpoint_url = s3_hook.conn_config.get_service_endpoint_url(
          service_name="s3"
      )
  
      config_kwargs: dict[
          str, Any
      ] = s3_hook.conn_config.extra_config.get("config_kwargs", {})
      config_kwargs.update(storage_options or {})
  
      register_events: dict[str, Callable[[Properties], None]] = {}
  
      s3_service_config = s3_hook.service_config
      if signer := s3_service_config.get("signer", None):
          log.info("Loading signer %s", signer)
          if singer_func := SIGNERS.get(signer):
              uri = s3_service_config.get("signer_uri", None)
              token = s3_service_config.get("signer_token", None)
              if not uri or not token:
                  raise ValueError(
                      f"Signer {signer} requires uri and token"
                  )
  
              properties: Properties = {
                  "uri": uri,
                  "token": uri,
              }
              singer_func_with_properties = partial(
                  singer_func, properties
              )
              register_events[
                  "before-sign.s3"
              ] = singer_func_with_properties
  
              # Disable the AWS Signer
              config_kwargs["signature_version"] = UNSIGNED
          else:
              raise ValueError(f"Signer not available: {signer}")
  
      if proxy_uri := s3_service_config.get(S3_PROXY_URI, None):
          config_kwargs["proxies"] = {
              "http": proxy_uri,
              "https": proxy_uri,
          }
  
      anon = False
      if asyncio.run(session.get_credentials()) is None:
          log.info("No credentials found, using anonymous access")
          anon = True
  
      fs = S3FileSystem(
          session=session,
          config_kwargs=config_kwargs,
          endpoint_url=endpoint_url,
          anon=anon,
      )
  
      for event_name, event_function in register_events.items():
          fs.s3.meta.events.register_last(
              event_name, event_function, unique_id=1925
          )
  
      return fs
  
file_path: airflow/airflow/providers/amazon/aws/fs/s3.py
imports: from __future__ import annotations
  import asyncio
  import logging
  from functools import partial
  from typing import TYPE_CHECKING, Any, Callable, Dict
  import requests
  from botocore import UNSIGNED
  from requests import HTTPError
  from airflow.providers.amazon.aws.hooks.s3 import S3Hook
label_desc: Use image node_id 1 for calling a global function with example usage: get_fs(conn_id, storage_options) and returns: fs