code: def get_fs(
      scheme: str,
      conn_id: str | None = None,
      storage_options: Properties | None = None,
  ) -> AbstractFileSystem:
      """
      Get a filesystem by scheme.
  
      :param scheme: the scheme to get the filesystem for
      :return: the filesystem method
      :param conn_id: the airflow connection id to use
      :param storage_options: the storage options to pass to the filesystem
      """
      filesystems = _register_filesystems()
      try:
          fs = filesystems[scheme]
      except KeyError:
          raise ValueError(
              f"No filesystem registered for scheme {scheme}"
          ) from None
  
      options = storage_options or {}
  
      # MyPy does not recognize dynamic parameters inspection when we call the method, and we have to do
      # it for compatibility reasons with already released providers, that's why we need to ignore
      # mypy errors here
      parameters = inspect.signature(fs).parameters
      if len(parameters) == 1:
          if options:
              raise AttributeError(
                  f"Filesystem {scheme} does not support storage options, but options were passed."
                  f"This most likely means that you are using an old version of the provider that does not "
                  f"support storage options. Please upgrade the provider if possible."
              )
          return fs(conn_id)  # type: ignore[call-arg]
      return fs(conn_id, options)
  
file_path: airflow/airflow/io/__init__.py
imports: from __future__ import annotations
  import inspect
  import logging
  from typing import TYPE_CHECKING, Callable, Mapping
  from fsspec.implementations.local import LocalFileSystem
  from airflow.compat.functools import cache
  from airflow.providers_manager import ProvidersManager
  from airflow.stats import Stats
  from airflow.utils.module_loading import import_string
label_desc: Use image node_id 3 for calling a global function with example usage: get_fs(scheme, conn_id, storage_options) and returns: fs, fs