prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Address warning in Airflow plugin version check
  > not super urgent, but as we move into supporting newer Python versions 
  > (https://github.com/meltano/meltano/pull/6135) and bumping Meltano's dependencies (https://github.com/meltano/meltano/issues/6264), we might break Airflow support.
  > 
  > It's also probably a very quick (< 1 hour) fix by replacing distutils.StrictVersion with the packaging.version equivalent.
  > 
  
  ```
  src/meltano/core/plugin/airflow.py:110: DeprecationWarning: distutils Version classes are deprecated. Use packaging.version instead.
      if StrictVersion(version) < StrictVersion("2.0.0")
  ```
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `src/meltano/core/plugin/airflow.py`
  Content:
  ```
  1 """Plugin glue code for Airflow."""
  2 import configparser
  3 import logging
  4 import os
  5 import subprocess
  6 from distutils.version import StrictVersion
  7 
  8 from meltano.core.behavior.hookable import hook
  9 from meltano.core.error import AsyncSubprocessError
  10 from meltano.core.plugin_invoker import PluginInvoker
  11 from meltano.core.utils import nest
  12 
  13 from . import BasePlugin, PluginType
  14 
  15 
  16 class AirflowInvoker(PluginInvoker):
  17     """Invoker that prepares env for Airflow."""
  18 
  19     def env(self):
  20         """Environment variables for Airflow.
  21 
  22         Returns:
  23             Dictionary of environment variables.
  24         """
  25         env = super().env()
  26 
  27         env["AIRFLOW_HOME"] = str(self.plugin_config_service.run_dir)
  28         env["AIRFLOW_CONFIG"] = str(self.files["config"])
  29 
  30         return env
  31 
  32 
  33 class Airflow(BasePlugin):
  34     """Plugin glue code for Airflow."""
  35 
  36     __plugin_type__ = PluginType.ORCHESTRATORS
  37 
  38     invoker_class = AirflowInvoker
  39 
  40     @property
  41     def config_files(self):
  42         """Return the configuration files required by the plugin.
  43 
  44         Returns:
  45             Dictionary of config file identifiers and filenames
  46         """
  47         return {"config": "airflow.cfg"}
  48 
  49     def process_config(self, flat_config):
  50         """Unflatten the config.
  51 
  52         Args:
  53             flat_config: the flat config
  54 
  55         Returns:
  56             unflattened config
  57         """
  58         config = {}
  59         for key, value in flat_config.items():
  60             nest(config, key, str(value))
  61         return config
  62 
  63     @staticmethod
  64     def update_config_file(invoker: AirflowInvoker) -> None:
  65         """Update airflow.cfg with plugin configuration.
  66 
  67         Args:
  68             invoker: the active PluginInvoker
  69         """
  70         airflow_cfg_path = invoker.files["config"]
  71         logging.debug(f"Generated default '{str(airflow_cfg_path)}'")
  72 
  73         # open the configuration and update it
  74         # now we let's update the config to use our stubs
  75         airflow_cfg = configparser.ConfigParser()
  76 
  77         with airflow_cfg_path.open() as airflow_cfg_file_to_read:
  78             airflow_cfg.read_file(airflow_cfg_file_to_read)
  79             logging.debug(f"Loaded '{str(airflow_cfg_path)}'")
  80 
  81         config = invoker.plugin_config_processed
  82         for section, section_config in config.items():
  83             airflow_cfg[section].update(section_config)
  84             logging.debug(f"\tUpdated section [{section}] with {section_config}")
  85 
  86         with airflow_cfg_path.open("w") as airflow_cfg_file_to_write:
  87             airflow_cfg.write(airflow_cfg_file_to_write)
  88             logging.debug(f"Saved '{str(airflow_cfg_path)}'")
  89 
  90     @hook("before_install")
  91     async def setup_env(self, *args, **kwargs):
  92         """Configure the env to make airflow installable without GPL deps.
  93 
  94         Args:
  95             args: Arbitrary args
  96             kwargs: Arbitrary kwargs
  97         """
  98         os.environ["SLUGIFY_USES_TEXT_UNIDECODE"] = "yes"
  99 
  100     @hook("before_configure")
  101     async def before_configure(self, invoker: AirflowInvoker, session):  # noqa: WPS217
  102         """Generate config file and keep metadata database up-to-date.
  103 
  104         Args:
  105             invoker: the active PluginInvoker
  106             session: metadata database session
  107 
  108         Raises:
  109             AsyncSubprocessError: if command failed to run
  110         """
  111         # generate the default `airflow.cfg`
  112         handle = await invoker.invoke_async(
  113             "--help",
  114             require_preparation=False,
  115             stdout=subprocess.DEVNULL,
  116             stderr=subprocess.PIPE,
  117         )
  118         exit_code = await handle.wait()
  119 
  120         if exit_code:
  121             raise AsyncSubprocessError(
  122                 "Command `airflow --help` failed", process=handle
  123             )
  124 
  125         # Read and update airflow.cfg
  126         self.update_config_file(invoker)
  127 
  128         # we've changed the configuration here, so we need to call
  129         # prepare again on the invoker so it re-reads the configuration
  130         # for the Airflow plugin
  131         await invoker.prepare(session)
  132 
  133         # make sure we use correct db init
  134         handle = await invoker.invoke_async(
  135             "version",
  136             stdout=subprocess.PIPE,
  137             stderr=subprocess.PIPE,
  138         )
  139 
  140         stdout, stderr = await handle.communicate()
  141 
  142         if handle.returncode:
  143             raise AsyncSubprocessError(
  144                 "Command `airflow version` failed", process=handle
  145             )
  146 
  147         version = stdout.decode()
  148         init_db_cmd = (
  149             ["initdb"]
  150             if StrictVersion(version) < StrictVersion("2.0.0")
  151             else ["db", "init"]
  152         )
  153 
  154         handle = await invoker.invoke_async(
  155             *init_db_cmd,
  156             stdout=subprocess.PIPE,
  157             stderr=subprocess.PIPE,
  158         )
  159         exit_code = await handle.wait()
  160 
  161         if exit_code:
  162             raise AsyncSubprocessError(
  163                 "Airflow metadata database could not be initialized: `airflow initdb` failed",
  164                 handle,
  165             )
  166 
  167         logging.debug("Completed `airflow initdb`")
  168 
  169     @hook("before_cleanup")
  170     async def before_cleanup(self, invoker: PluginInvoker):
  171         """Delete the config file.
  172 
  173         Args:
  174             invoker: the active PluginInvoker
  175         """
  176         config_file = invoker.files["config"]
  177         try:
  178             config_file.unlink()
  179             logging.debug(f"Deleted configuration at {config_file}")
  180         except FileNotFoundError:
  181             pass
  182
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/src/meltano/core/plugin/airflow.py b/src/meltano/core/plugin/airflow.py
  --- a/src/meltano/core/plugin/airflow.py
  +++ b/src/meltano/core/plugin/airflow.py
  @@ -3,7 +3,8 @@
   import logging
   import os
   import subprocess
  -from distutils.version import StrictVersion
  +
  +from packaging.version import Version
   
   from meltano.core.behavior.hookable import hook
   from meltano.core.error import AsyncSubprocessError
  @@ -146,9 +147,7 @@
   
           version = stdout.decode()
           init_db_cmd = (
  -            ["initdb"]
  -            if StrictVersion(version) < StrictVersion("2.0.0")
  -            else ["db", "init"]
  +            ["initdb"] if Version(version) < Version("2.0.0") else ["db", "init"]
           )
   
           handle = await invoker.invoke_async(
  
verification_info: {"golden_diff": "diff --git a/src/meltano/core/plugin/airflow.py b/src/meltano/core/plugin/airflow.py\n--- a/src/meltano/core/plugin/airflow.py\n+++ b/src/meltano/core/plugin/airflow.py\n@@ -3,7 +3,8 @@\n import logging\n import os\n import subprocess\n-from distutils.version import StrictVersion\n+\n+from packaging.version import Version\n \n from meltano.core.behavior.hookable import hook\n from meltano.core.error import AsyncSubprocessError\n@@ -146,9 +147,7 @@\n \n         version = stdout.decode()\n         init_db_cmd = (\n-            [\"initdb\"]\n-            if StrictVersion(version) < StrictVersion(\"2.0.0\")\n-            else [\"db\", \"init\"]\n+            [\"initdb\"] if Version(version) < Version(\"2.0.0\") else [\"db\", \"init\"]\n         )\n \n         handle = await invoker.invoke_async(\n", "issue": "Address warning in Airflow plugin version check\n> not super urgent, but as we move into supporting newer Python versions \r\n> (https://github.com/meltano/meltano/pull/6135) and bumping Meltano's dependencies (https://github.com/meltano/meltano/issues/6264), we might break Airflow support.\r\n> \r\n> It's also probably a very quick (< 1 hour) fix by replacing distutils.StrictVersion with the packaging.version equivalent.\r\n> \r\n\r\n```\r\nsrc/meltano/core/plugin/airflow.py:110: DeprecationWarning: distutils Version classes are deprecated. Use packaging.version instead.\r\n    if StrictVersion(version) < StrictVersion(\"2.0.0\")\r\n```\n", "before_files": [{"content": "\"\"\"Plugin glue code for Airflow.\"\"\"\nimport configparser\nimport logging\nimport os\nimport subprocess\nfrom distutils.version import StrictVersion\n\nfrom meltano.core.behavior.hookable import hook\nfrom meltano.core.error import AsyncSubprocessError\nfrom meltano.core.plugin_invoker import PluginInvoker\nfrom meltano.core.utils import nest\n\nfrom . import BasePlugin, PluginType\n\n\nclass AirflowInvoker(PluginInvoker):\n    \"\"\"Invoker that prepares env for Airflow.\"\"\"\n\n    def env(self):\n        \"\"\"Environment variables for Airflow.\n\n        Returns:\n            Dictionary of environment variables.\n        \"\"\"\n        env = super().env()\n\n        env[\"AIRFLOW_HOME\"] = str(self.plugin_config_service.run_dir)\n        env[\"AIRFLOW_CONFIG\"] = str(self.files[\"config\"])\n\n        return env\n\n\nclass Airflow(BasePlugin):\n    \"\"\"Plugin glue code for Airflow.\"\"\"\n\n    __plugin_type__ = PluginType.ORCHESTRATORS\n\n    invoker_class = AirflowInvoker\n\n    @property\n    def config_files(self):\n        \"\"\"Return the configuration files required by the plugin.\n\n        Returns:\n            Dictionary of config file identifiers and filenames\n        \"\"\"\n        return {\"config\": \"airflow.cfg\"}\n\n    def process_config(self, flat_config):\n        \"\"\"Unflatten the config.\n\n        Args:\n            flat_config: the flat config\n\n        Returns:\n            unflattened config\n        \"\"\"\n        config = {}\n        for key, value in flat_config.items():\n            nest(config, key, str(value))\n        return config\n\n    @staticmethod\n    def update_config_file(invoker: AirflowInvoker) -> None:\n        \"\"\"Update airflow.cfg with plugin configuration.\n\n        Args:\n            invoker: the active PluginInvoker\n        \"\"\"\n        airflow_cfg_path = invoker.files[\"config\"]\n        logging.debug(f\"Generated default '{str(airflow_cfg_path)}'\")\n\n        # open the configuration and update it\n        # now we let's update the config to use our stubs\n        airflow_cfg = configparser.ConfigParser()\n\n        with airflow_cfg_path.open() as airflow_cfg_file_to_read:\n            airflow_cfg.read_file(airflow_cfg_file_to_read)\n            logging.debug(f\"Loaded '{str(airflow_cfg_path)}'\")\n\n        config = invoker.plugin_config_processed\n        for section, section_config in config.items():\n            airflow_cfg[section].update(section_config)\n            logging.debug(f\"\\tUpdated section [{section}] with {section_config}\")\n\n        with airflow_cfg_path.open(\"w\") as airflow_cfg_file_to_write:\n            airflow_cfg.write(airflow_cfg_file_to_write)\n            logging.debug(f\"Saved '{str(airflow_cfg_path)}'\")\n\n    @hook(\"before_install\")\n    async def setup_env(self, *args, **kwargs):\n        \"\"\"Configure the env to make airflow installable without GPL deps.\n\n        Args:\n            args: Arbitrary args\n            kwargs: Arbitrary kwargs\n        \"\"\"\n        os.environ[\"SLUGIFY_USES_TEXT_UNIDECODE\"] = \"yes\"\n\n    @hook(\"before_configure\")\n    async def before_configure(self, invoker: AirflowInvoker, session):  # noqa: WPS217\n        \"\"\"Generate config file and keep metadata database up-to-date.\n\n        Args:\n            invoker: the active PluginInvoker\n            session: metadata database session\n\n        Raises:\n            AsyncSubprocessError: if command failed to run\n        \"\"\"\n        # generate the default `airflow.cfg`\n        handle = await invoker.invoke_async(\n            \"--help\",\n            require_preparation=False,\n            stdout=subprocess.DEVNULL,\n            stderr=subprocess.PIPE,\n        )\n        exit_code = await handle.wait()\n\n        if exit_code:\n            raise AsyncSubprocessError(\n                \"Command `airflow --help` failed\", process=handle\n            )\n\n        # Read and update airflow.cfg\n        self.update_config_file(invoker)\n\n        # we've changed the configuration here, so we need to call\n        # prepare again on the invoker so it re-reads the configuration\n        # for the Airflow plugin\n        await invoker.prepare(session)\n\n        # make sure we use correct db init\n        handle = await invoker.invoke_async(\n            \"version\",\n            stdout=subprocess.PIPE,\n            stderr=subprocess.PIPE,\n        )\n\n        stdout, stderr = await handle.communicate()\n\n        if handle.returncode:\n            raise AsyncSubprocessError(\n                \"Command `airflow version` failed\", process=handle\n            )\n\n        version = stdout.decode()\n        init_db_cmd = (\n            [\"initdb\"]\n            if StrictVersion(version) < StrictVersion(\"2.0.0\")\n            else [\"db\", \"init\"]\n        )\n\n        handle = await invoker.invoke_async(\n            *init_db_cmd,\n            stdout=subprocess.PIPE,\n            stderr=subprocess.PIPE,\n        )\n        exit_code = await handle.wait()\n\n        if exit_code:\n            raise AsyncSubprocessError(\n                \"Airflow metadata database could not be initialized: `airflow initdb` failed\",\n                handle,\n            )\n\n        logging.debug(\"Completed `airflow initdb`\")\n\n    @hook(\"before_cleanup\")\n    async def before_cleanup(self, invoker: PluginInvoker):\n        \"\"\"Delete the config file.\n\n        Args:\n            invoker: the active PluginInvoker\n        \"\"\"\n        config_file = invoker.files[\"config\"]\n        try:\n            config_file.unlink()\n            logging.debug(f\"Deleted configuration at {config_file}\")\n        except FileNotFoundError:\n            pass\n", "path": "src/meltano/core/plugin/airflow.py"}], "after_files": [{"content": "\"\"\"Plugin glue code for Airflow.\"\"\"\nimport configparser\nimport logging\nimport os\nimport subprocess\n\nfrom packaging.version import Version\n\nfrom meltano.core.behavior.hookable import hook\nfrom meltano.core.error import AsyncSubprocessError\nfrom meltano.core.plugin_invoker import PluginInvoker\nfrom meltano.core.utils import nest\n\nfrom . import BasePlugin, PluginType\n\n\nclass AirflowInvoker(PluginInvoker):\n    \"\"\"Invoker that prepares env for Airflow.\"\"\"\n\n    def env(self):\n        \"\"\"Environment variables for Airflow.\n\n        Returns:\n            Dictionary of environment variables.\n        \"\"\"\n        env = super().env()\n\n        env[\"AIRFLOW_HOME\"] = str(self.plugin_config_service.run_dir)\n        env[\"AIRFLOW_CONFIG\"] = str(self.files[\"config\"])\n\n        return env\n\n\nclass Airflow(BasePlugin):\n    \"\"\"Plugin glue code for Airflow.\"\"\"\n\n    __plugin_type__ = PluginType.ORCHESTRATORS\n\n    invoker_class = AirflowInvoker\n\n    @property\n    def config_files(self):\n        \"\"\"Return the configuration files required by the plugin.\n\n        Returns:\n            Dictionary of config file identifiers and filenames\n        \"\"\"\n        return {\"config\": \"airflow.cfg\"}\n\n    def process_config(self, flat_config):\n        \"\"\"Unflatten the config.\n\n        Args:\n            flat_config: the flat config\n\n        Returns:\n            unflattened config\n        \"\"\"\n        config = {}\n        for key, value in flat_config.items():\n            nest(config, key, str(value))\n        return config\n\n    @staticmethod\n    def update_config_file(invoker: AirflowInvoker) -> None:\n        \"\"\"Update airflow.cfg with plugin configuration.\n\n        Args:\n            invoker: the active PluginInvoker\n        \"\"\"\n        airflow_cfg_path = invoker.files[\"config\"]\n        logging.debug(f\"Generated default '{str(airflow_cfg_path)}'\")\n\n        # open the configuration and update it\n        # now we let's update the config to use our stubs\n        airflow_cfg = configparser.ConfigParser()\n\n        with airflow_cfg_path.open() as airflow_cfg_file_to_read:\n            airflow_cfg.read_file(airflow_cfg_file_to_read)\n            logging.debug(f\"Loaded '{str(airflow_cfg_path)}'\")\n\n        config = invoker.plugin_config_processed\n        for section, section_config in config.items():\n            airflow_cfg[section].update(section_config)\n            logging.debug(f\"\\tUpdated section [{section}] with {section_config}\")\n\n        with airflow_cfg_path.open(\"w\") as airflow_cfg_file_to_write:\n            airflow_cfg.write(airflow_cfg_file_to_write)\n            logging.debug(f\"Saved '{str(airflow_cfg_path)}'\")\n\n    @hook(\"before_install\")\n    async def setup_env(self, *args, **kwargs):\n        \"\"\"Configure the env to make airflow installable without GPL deps.\n\n        Args:\n            args: Arbitrary args\n            kwargs: Arbitrary kwargs\n        \"\"\"\n        os.environ[\"SLUGIFY_USES_TEXT_UNIDECODE\"] = \"yes\"\n\n    @hook(\"before_configure\")\n    async def before_configure(self, invoker: AirflowInvoker, session):  # noqa: WPS217\n        \"\"\"Generate config file and keep metadata database up-to-date.\n\n        Args:\n            invoker: the active PluginInvoker\n            session: metadata database session\n\n        Raises:\n            AsyncSubprocessError: if command failed to run\n        \"\"\"\n        # generate the default `airflow.cfg`\n        handle = await invoker.invoke_async(\n            \"--help\",\n            require_preparation=False,\n            stdout=subprocess.DEVNULL,\n            stderr=subprocess.PIPE,\n        )\n        exit_code = await handle.wait()\n\n        if exit_code:\n            raise AsyncSubprocessError(\n                \"Command `airflow --help` failed\", process=handle\n            )\n\n        # Read and update airflow.cfg\n        self.update_config_file(invoker)\n\n        # we've changed the configuration here, so we need to call\n        # prepare again on the invoker so it re-reads the configuration\n        # for the Airflow plugin\n        await invoker.prepare(session)\n\n        # make sure we use correct db init\n        handle = await invoker.invoke_async(\n            \"version\",\n            stdout=subprocess.PIPE,\n            stderr=subprocess.PIPE,\n        )\n\n        stdout, stderr = await handle.communicate()\n\n        if handle.returncode:\n            raise AsyncSubprocessError(\n                \"Command `airflow version` failed\", process=handle\n            )\n\n        version = stdout.decode()\n        init_db_cmd = (\n            [\"initdb\"] if Version(version) < Version(\"2.0.0\") else [\"db\", \"init\"]\n        )\n\n        handle = await invoker.invoke_async(\n            *init_db_cmd,\n            stdout=subprocess.PIPE,\n            stderr=subprocess.PIPE,\n        )\n        exit_code = await handle.wait()\n\n        if exit_code:\n            raise AsyncSubprocessError(\n                \"Airflow metadata database could not be initialized: `airflow initdb` failed\",\n                handle,\n            )\n\n        logging.debug(\"Completed `airflow initdb`\")\n\n    @hook(\"before_cleanup\")\n    async def before_cleanup(self, invoker: PluginInvoker):\n        \"\"\"Delete the config file.\n\n        Args:\n            invoker: the active PluginInvoker\n        \"\"\"\n        config_file = invoker.files[\"config\"]\n        try:\n            config_file.unlink()\n            logging.debug(f\"Deleted configuration at {config_file}\")\n        except FileNotFoundError:\n            pass\n", "path": "src/meltano/core/plugin/airflow.py"}]}