We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
bug: `meltano state list` with pattern - no such option
### Meltano Version

2.3.0

### Python Version

3.8

### Bug scope

CLI (options, error messages, logging, etc.)

### Operating System

Mac

### Description

It looks like the `--pattern` argument thats in the docs https://docs.meltano.com/reference/command-line-interface#list isnt available on the CLI.

```
(meltano) Patricks-MBP:data pnadolny$ meltano --version
meltano, version 2.3.0

(meltano) Patricks-MBP:data pnadolny$ meltano state list --pattern '*tap-gitlab*'
2022-07-25T21:31:25.438941Z [info     ] Environment 'userdev' is active
Usage: meltano state list [OPTIONS] [PATTERN]
Try 'meltano state list --help' for help.

Error: No such option: --pattern
```

### Code

_No response_

--- 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/cli/state.py`
Content:
```
1 """State management in CLI."""
2 from __future__ import annotations
3 
4 import json
5 import re
6 from datetime import datetime as dt
7 from functools import partial, reduce, wraps
8 from operator import xor
9 
10 import click
11 import structlog
12 
13 from meltano.cli.params import pass_project
14 from meltano.core.block.parser import BlockParser
15 from meltano.core.db import project_engine
16 from meltano.core.job import Payload
17 from meltano.core.project import Project
18 from meltano.core.state_service import InvalidJobStateError, StateService
19 
20 from . import cli
21 from .utils import InstrumentedCmd, InstrumentedGroup
22 
23 STATE_SERVICE_KEY = "state_service"
24 
25 logger = structlog.getLogger(__name__)
26 
27 
28 class MutuallyExclusiveOptionsError(Exception):
29     """Occurs when mutually exclusive options are provided incorrectly."""
30 
31     def __init__(self, *options: str) -> None:
32         """Instantiate the error.
33 
34         Args:
35             options: the mutually exclusive options that were incorrectly provided.
36         """
37         super().__init__(*options)
38         self.options = options
39 
40     def __str__(self) -> str:
41         """Represent the error as a string."""
42         return f"Must provide exactly one of: {','.join(self.options)}"
43 
44 
45 def _prompt_for_confirmation(prompt):
46     """Wrap destructive CLI commands which should prompt the user for confirmation."""
47 
48     def wrapper(func):
49         fun = click.option(
50             "--force", is_flag=True, help="Don't prompt for confirmation."
51         )(func)
52 
53         @wraps(func)
54         def _wrapper(force=False, *args, **kwargs):
55             if force or click.confirm(prompt):
56                 return fun(*args, **kwargs, force=force)
57             else:
58                 click.secho("Aborting.", fg="red")
59 
60         return _wrapper
61 
62     return wrapper
63 
64 
65 prompt_for_confirmation = partial(
66     _prompt_for_confirmation, prompt="This is a destructive command. Continue?"
67 )
68 
69 
70 def state_service_from_state_id(project: Project, state_id: str) -> StateService | None:
71     """Instantiate by parsing a state_id."""
72     state_id_re = re.compile(r"^(?P<env>.+)\:(?P<tap>.+)-to-(?P<target>.+)$")
73     match = state_id_re.match(state_id)
74     if match:
75         # If the state_id matches convention (i.e., job has been run via "meltano run"),
76         # try parsing into BlockSet.
77         # This way, we get BlockSet validation and raise an error if no
78         # plugin in the BlockSet has "state" capability
79         try:
80             if not project.active_environment:
81                 logger.warn(
82                     f"Running state operation for environment '{match.group('env')}' outside of an environment"
83                 )
84             elif project.active_environment.name != match.group("env"):
85                 logger.warn(
86                     f"Environment '{match.group('env')}' used in state operation does not match current environment '{project.active_environment.name}'."
87                 )
88             project.activate_environment(match.group("env"))
89             blocks = [match.group("tap"), match.group("target")]
90             parser = BlockParser(logger, project, blocks)
91             return next(parser.find_blocks()).state_service
92         except Exception:
93             logger.warn("No plugins found for provided state_id.")
94     # If provided state_id does not match convention (i.e., run via "meltano elt"),
95     # use the standalone StateService in the CLI context.
96     return None
97 
98 
99 @cli.group(cls=InstrumentedGroup, name="state", short_help="Manage Singer state.")
100 @click.pass_context
101 @pass_project(migrate=True)
102 def meltano_state(project: Project, ctx: click.Context):
103     """
104     Manage state.
105 
106     \b\nRead more at https://docs.meltano.com/reference/command-line-interface#state
107     """
108     _, sessionmaker = project_engine(project)
109     session = sessionmaker()
110     ctx.obj[STATE_SERVICE_KEY] = StateService(session)  # noqa: WPS204
111 
112 
113 @meltano_state.command(cls=InstrumentedCmd, name="list")
114 @click.argument("pattern", required=False)
115 @click.pass_context
116 @pass_project()
117 def list_state(
118     project: Project, ctx: click.Context, pattern: str | None
119 ):  # noqa: WPS125
120     """List all state_ids for this project.
121 
122     Optionally pass a glob-style pattern to filter state_ids by.
123     """
124     state_service = ctx.obj[STATE_SERVICE_KEY]
125     ctx.obj["legacy_tracker"].track_meltano_state("list")
126     states = state_service.list_state(pattern)
127     if states:
128         for state_id, state in states.items():
129             if state:
130                 try:
131                     state_service.validate_state(json.dumps(state))
132                 except (InvalidJobStateError, json.decoder.JSONDecodeError):
133                     click.secho(state_id, fg="red")
134                 else:
135                     click.secho(state_id, fg="green")
136             else:
137                 click.secho(state_id, fg="yellow")
138     else:
139         logger.info("No state IDs found.")
140 
141 
142 @meltano_state.command(cls=InstrumentedCmd, name="copy")
143 @prompt_for_confirmation(
144     prompt="This will overwrite state for the destination. Continue?"
145 )
146 @click.argument("src-state-id", type=str)
147 @click.argument("dst-state-id", type=str)
148 @pass_project(migrate=True)
149 @click.pass_context
150 def copy_state(
151     ctx: click.Context,
152     project: Project,
153     src_state_id: str,
154     dst_state_id: str,
155     force: bool,
156 ):
157     """Copy state to another job id."""
158     # Retrieve state for copying
159     state_service = (
160         state_service_from_state_id(project, src_state_id) or ctx.obj[STATE_SERVICE_KEY]
161     )
162     ctx.obj["legacy_tracker"].track_meltano_state("copy", dst_state_id)
163 
164     state_service.copy_state(src_state_id, dst_state_id)
165 
166     logger.info(
167         f"State for {dst_state_id} was successfully copied from {src_state_id} at {dt.utcnow():%Y-%m-%d %H:%M:%S}."  # noqa: WPS323
168     )
169 
170 
171 @meltano_state.command(cls=InstrumentedCmd, name="move")
172 @prompt_for_confirmation(
173     prompt="This will clear the source state and overwrite destination state. Continue?"
174 )
175 @click.argument("src-state-id", type=str)
176 @click.argument("dst-state-id", type=str)
177 @pass_project(migrate=True)
178 @click.pass_context
179 def move_state(
180     ctx: click.Context,
181     project: Project,
182     src_state_id: str,
183     dst_state_id: str,
184     force: bool,
185 ):
186     """Move state to another job id, clearing the original."""
187     # Retrieve state for moveing
188     state_service = (
189         state_service_from_state_id(project, dst_state_id) or ctx.obj[STATE_SERVICE_KEY]
190     )
191     ctx.obj["legacy_tracker"].track_meltano_state("move", dst_state_id)
192 
193     state_service.move_state(src_state_id, dst_state_id)
194 
195     logger.info(
196         f"State for {src_state_id} was successfully moved to {dst_state_id} at {dt.utcnow():%Y-%m-%d %H:%M:%S}."  # noqa: WPS323
197     )
198 
199 
200 @meltano_state.command(cls=InstrumentedCmd, name="merge")
201 @click.option(
202     "--from-state-id",
203     type=str,
204     help="Merge state from an existing state ID.",
205 )
206 @click.option(
207     "--input-file",
208     type=click.Path(exists=True),
209     help="Merge state from a JSON file containing Singer state.",
210 )
211 @click.argument("state-id", type=str)
212 @click.argument("state", type=str, required=False)
213 @pass_project(migrate=True)
214 @click.pass_context
215 def merge_state(
216     ctx: click.Context,
217     project: Project,
218     state_id: str,
219     state: str | None,
220     input_file: click.Path | None,
221     from_state_id: str | None,
222 ):
223     """Add bookmarks to existing state."""
224     state_service = (
225         state_service_from_state_id(project, state_id) or ctx.obj[STATE_SERVICE_KEY]
226     )
227     ctx.obj["legacy_tracker"].track_meltano_state("merge", state_id)
228     mutually_exclusive_options = ["--input-file", "STATE", "--from-state-id"]
229     if not reduce(xor, map(bool, [state, input_file, from_state_id])):
230         raise MutuallyExclusiveOptionsError(*mutually_exclusive_options)
231     elif input_file:
232         with open(input_file) as state_f:
233             state_service.add_state(
234                 state_id, state_f.read(), payload_flags=Payload.INCOMPLETE_STATE
235             )
236     elif state:
237         state_service.add_state(state_id, state, payload_flags=Payload.INCOMPLETE_STATE)
238     elif from_state_id:
239         state_service.merge_state(from_state_id, state_id)
240     logger.info(
241         f"State for {state_id} was successfully merged at {dt.utcnow():%Y-%m-%d %H:%M:%S}."  # noqa: WPS323
242     )
243 
244 
245 @meltano_state.command(cls=InstrumentedCmd, name="set")
246 @prompt_for_confirmation(
247     prompt="This will overwrite the state's current value. Continue?"
248 )
249 @click.option(
250     "--input-file",
251     type=click.Path(exists=True),
252     help="Set state from json file containing Singer state.",
253 )
254 @click.argument("state-id")
255 @click.argument("state", type=str, required=False)
256 @pass_project(migrate=True)
257 @click.pass_context
258 def set_state(
259     ctx: click.Context,
260     project: Project,
261     state_id: str,
262     state: str | None,
263     input_file: click.Path | None,
264     force: bool,
265 ):
266     """Set state."""
267     state_service = (
268         state_service_from_state_id(project, state_id) or ctx.obj[STATE_SERVICE_KEY]
269     )
270     ctx.obj["legacy_tracker"].track_meltano_state("set", state_id)
271     if not reduce(xor, map(bool, [state, input_file])):
272         raise MutuallyExclusiveOptionsError("--input-file", "STATE")
273     elif input_file:
274         with open(input_file) as state_f:
275             state_service.set_state(state_id, state_f.read())
276     elif state:
277         state_service.set_state(state_id, state)
278     logger.info(
279         f"State for {state_id} was successfully set at {dt.utcnow():%Y-%m-%d %H:%M:%S}."  # noqa: WPS323
280     )
281 
282 
283 @meltano_state.command(cls=InstrumentedCmd, name="get")  # noqa: WPS46
284 @click.argument("state-id")
285 @pass_project(migrate=True)
286 @click.pass_context
287 def get_state(ctx: click.Context, project: Project, state_id: str):  # noqa: WPS463
288     """Get state."""
289     state_service = (
290         state_service_from_state_id(project, state_id) or ctx.obj[STATE_SERVICE_KEY]
291     )
292     ctx.obj["legacy_tracker"].track_meltano_state("get", state_id)
293     retrieved_state = state_service.get_state(state_id)
294     click.echo(json.dumps(retrieved_state))
295 
296 
297 @meltano_state.command(cls=InstrumentedCmd, name="clear")
298 @prompt_for_confirmation(prompt="This will clear state for the job. Continue?")
299 @click.argument("state-id")
300 @pass_project(migrate=True)
301 @click.pass_context
302 def clear_state(ctx: click.Context, project: Project, state_id: str, force: bool):
303     """Clear state."""
304     state_service = (
305         state_service_from_state_id(project, state_id) or ctx.obj[STATE_SERVICE_KEY]
306     )
307     ctx.obj["legacy_tracker"].track_meltano_state("clear", state_id)
308     state_service.clear_state(state_id)
309
```
--- 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()

```

diff --git a/src/meltano/cli/state.py b/src/meltano/cli/state.py
--- a/src/meltano/cli/state.py
+++ b/src/meltano/cli/state.py
@@ -111,7 +111,7 @@
 
 
 @meltano_state.command(cls=InstrumentedCmd, name="list")
-@click.argument("pattern", required=False)
+@click.option("--pattern", type=str, help="Filter state IDs by pattern.")
 @click.pass_context
 @pass_project()
 def list_state(

{"golden_diff": "diff --git a/src/meltano/cli/state.py b/src/meltano/cli/state.py\n--- a/src/meltano/cli/state.py\n+++ b/src/meltano/cli/state.py\n@@ -111,7 +111,7 @@\n \n \n @meltano_state.command(cls=InstrumentedCmd, name=\"list\")\n-@click.argument(\"pattern\", required=False)\n+@click.option(\"--pattern\", type=str, help=\"Filter state IDs by pattern.\")\n @click.pass_context\n @pass_project()\n def list_state(\n", "issue": "bug: `meltano state list` with pattern - no such option\n### Meltano Version\r\n\r\n2.3.0\r\n\r\n### Python Version\r\n\r\n3.8\r\n\r\n### Bug scope\r\n\r\nCLI (options, error messages, logging, etc.)\r\n\r\n### Operating System\r\n\r\nMac\r\n\r\n### Description\r\n\r\nIt looks like the `--pattern` argument thats in the docs https://docs.meltano.com/reference/command-line-interface#list isnt available on the CLI.\r\n\r\n```\r\n(meltano) Patricks-MBP:data pnadolny$ meltano --version\r\nmeltano, version 2.3.0\r\n\r\n(meltano) Patricks-MBP:data pnadolny$ meltano state list --pattern '*tap-gitlab*'\r\n2022-07-25T21:31:25.438941Z [info     ] Environment 'userdev' is active\r\nUsage: meltano state list [OPTIONS] [PATTERN]\r\nTry 'meltano state list --help' for help.\r\n\r\nError: No such option: --pattern\r\n```\r\n\r\n### Code\r\n\r\n_No response_\n", "before_files": [{"content": "\"\"\"State management in CLI.\"\"\"\nfrom __future__ import annotations\n\nimport json\nimport re\nfrom datetime import datetime as dt\nfrom functools import partial, reduce, wraps\nfrom operator import xor\n\nimport click\nimport structlog\n\nfrom meltano.cli.params import pass_project\nfrom meltano.core.block.parser import BlockParser\nfrom meltano.core.db import project_engine\nfrom meltano.core.job import Payload\nfrom meltano.core.project import Project\nfrom meltano.core.state_service import InvalidJobStateError, StateService\n\nfrom . import cli\nfrom .utils import InstrumentedCmd, InstrumentedGroup\n\nSTATE_SERVICE_KEY = \"state_service\"\n\nlogger = structlog.getLogger(__name__)\n\n\nclass MutuallyExclusiveOptionsError(Exception):\n    \"\"\"Occurs when mutually exclusive options are provided incorrectly.\"\"\"\n\n    def __init__(self, *options: str) -> None:\n        \"\"\"Instantiate the error.\n\n        Args:\n            options: the mutually exclusive options that were incorrectly provided.\n        \"\"\"\n        super().__init__(*options)\n        self.options = options\n\n    def __str__(self) -> str:\n        \"\"\"Represent the error as a string.\"\"\"\n        return f\"Must provide exactly one of: {','.join(self.options)}\"\n\n\ndef _prompt_for_confirmation(prompt):\n    \"\"\"Wrap destructive CLI commands which should prompt the user for confirmation.\"\"\"\n\n    def wrapper(func):\n        fun = click.option(\n            \"--force\", is_flag=True, help=\"Don't prompt for confirmation.\"\n        )(func)\n\n        @wraps(func)\n        def _wrapper(force=False, *args, **kwargs):\n            if force or click.confirm(prompt):\n                return fun(*args, **kwargs, force=force)\n            else:\n                click.secho(\"Aborting.\", fg=\"red\")\n\n        return _wrapper\n\n    return wrapper\n\n\nprompt_for_confirmation = partial(\n    _prompt_for_confirmation, prompt=\"This is a destructive command. Continue?\"\n)\n\n\ndef state_service_from_state_id(project: Project, state_id: str) -> StateService | None:\n    \"\"\"Instantiate by parsing a state_id.\"\"\"\n    state_id_re = re.compile(r\"^(?P<env>.+)\\:(?P<tap>.+)-to-(?P<target>.+)$\")\n    match = state_id_re.match(state_id)\n    if match:\n        # If the state_id matches convention (i.e., job has been run via \"meltano run\"),\n        # try parsing into BlockSet.\n        # This way, we get BlockSet validation and raise an error if no\n        # plugin in the BlockSet has \"state\" capability\n        try:\n            if not project.active_environment:\n                logger.warn(\n                    f\"Running state operation for environment '{match.group('env')}' outside of an environment\"\n                )\n            elif project.active_environment.name != match.group(\"env\"):\n                logger.warn(\n                    f\"Environment '{match.group('env')}' used in state operation does not match current environment '{project.active_environment.name}'.\"\n                )\n            project.activate_environment(match.group(\"env\"))\n            blocks = [match.group(\"tap\"), match.group(\"target\")]\n            parser = BlockParser(logger, project, blocks)\n            return next(parser.find_blocks()).state_service\n        except Exception:\n            logger.warn(\"No plugins found for provided state_id.\")\n    # If provided state_id does not match convention (i.e., run via \"meltano elt\"),\n    # use the standalone StateService in the CLI context.\n    return None\n\n\n@cli.group(cls=InstrumentedGroup, name=\"state\", short_help=\"Manage Singer state.\")\n@click.pass_context\n@pass_project(migrate=True)\ndef meltano_state(project: Project, ctx: click.Context):\n    \"\"\"\n    Manage state.\n\n    \\b\\nRead more at https://docs.meltano.com/reference/command-line-interface#state\n    \"\"\"\n    _, sessionmaker = project_engine(project)\n    session = sessionmaker()\n    ctx.obj[STATE_SERVICE_KEY] = StateService(session)  # noqa: WPS204\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"list\")\n@click.argument(\"pattern\", required=False)\n@click.pass_context\n@pass_project()\ndef list_state(\n    project: Project, ctx: click.Context, pattern: str | None\n):  # noqa: WPS125\n    \"\"\"List all state_ids for this project.\n\n    Optionally pass a glob-style pattern to filter state_ids by.\n    \"\"\"\n    state_service = ctx.obj[STATE_SERVICE_KEY]\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"list\")\n    states = state_service.list_state(pattern)\n    if states:\n        for state_id, state in states.items():\n            if state:\n                try:\n                    state_service.validate_state(json.dumps(state))\n                except (InvalidJobStateError, json.decoder.JSONDecodeError):\n                    click.secho(state_id, fg=\"red\")\n                else:\n                    click.secho(state_id, fg=\"green\")\n            else:\n                click.secho(state_id, fg=\"yellow\")\n    else:\n        logger.info(\"No state IDs found.\")\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"copy\")\n@prompt_for_confirmation(\n    prompt=\"This will overwrite state for the destination. Continue?\"\n)\n@click.argument(\"src-state-id\", type=str)\n@click.argument(\"dst-state-id\", type=str)\n@pass_project(migrate=True)\n@click.pass_context\ndef copy_state(\n    ctx: click.Context,\n    project: Project,\n    src_state_id: str,\n    dst_state_id: str,\n    force: bool,\n):\n    \"\"\"Copy state to another job id.\"\"\"\n    # Retrieve state for copying\n    state_service = (\n        state_service_from_state_id(project, src_state_id) or ctx.obj[STATE_SERVICE_KEY]\n    )\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"copy\", dst_state_id)\n\n    state_service.copy_state(src_state_id, dst_state_id)\n\n    logger.info(\n        f\"State for {dst_state_id} was successfully copied from {src_state_id} at {dt.utcnow():%Y-%m-%d %H:%M:%S}.\"  # noqa: WPS323\n    )\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"move\")\n@prompt_for_confirmation(\n    prompt=\"This will clear the source state and overwrite destination state. Continue?\"\n)\n@click.argument(\"src-state-id\", type=str)\n@click.argument(\"dst-state-id\", type=str)\n@pass_project(migrate=True)\n@click.pass_context\ndef move_state(\n    ctx: click.Context,\n    project: Project,\n    src_state_id: str,\n    dst_state_id: str,\n    force: bool,\n):\n    \"\"\"Move state to another job id, clearing the original.\"\"\"\n    # Retrieve state for moveing\n    state_service = (\n        state_service_from_state_id(project, dst_state_id) or ctx.obj[STATE_SERVICE_KEY]\n    )\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"move\", dst_state_id)\n\n    state_service.move_state(src_state_id, dst_state_id)\n\n    logger.info(\n        f\"State for {src_state_id} was successfully moved to {dst_state_id} at {dt.utcnow():%Y-%m-%d %H:%M:%S}.\"  # noqa: WPS323\n    )\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"merge\")\n@click.option(\n    \"--from-state-id\",\n    type=str,\n    help=\"Merge state from an existing state ID.\",\n)\n@click.option(\n    \"--input-file\",\n    type=click.Path(exists=True),\n    help=\"Merge state from a JSON file containing Singer state.\",\n)\n@click.argument(\"state-id\", type=str)\n@click.argument(\"state\", type=str, required=False)\n@pass_project(migrate=True)\n@click.pass_context\ndef merge_state(\n    ctx: click.Context,\n    project: Project,\n    state_id: str,\n    state: str | None,\n    input_file: click.Path | None,\n    from_state_id: str | None,\n):\n    \"\"\"Add bookmarks to existing state.\"\"\"\n    state_service = (\n        state_service_from_state_id(project, state_id) or ctx.obj[STATE_SERVICE_KEY]\n    )\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"merge\", state_id)\n    mutually_exclusive_options = [\"--input-file\", \"STATE\", \"--from-state-id\"]\n    if not reduce(xor, map(bool, [state, input_file, from_state_id])):\n        raise MutuallyExclusiveOptionsError(*mutually_exclusive_options)\n    elif input_file:\n        with open(input_file) as state_f:\n            state_service.add_state(\n                state_id, state_f.read(), payload_flags=Payload.INCOMPLETE_STATE\n            )\n    elif state:\n        state_service.add_state(state_id, state, payload_flags=Payload.INCOMPLETE_STATE)\n    elif from_state_id:\n        state_service.merge_state(from_state_id, state_id)\n    logger.info(\n        f\"State for {state_id} was successfully merged at {dt.utcnow():%Y-%m-%d %H:%M:%S}.\"  # noqa: WPS323\n    )\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"set\")\n@prompt_for_confirmation(\n    prompt=\"This will overwrite the state's current value. Continue?\"\n)\n@click.option(\n    \"--input-file\",\n    type=click.Path(exists=True),\n    help=\"Set state from json file containing Singer state.\",\n)\n@click.argument(\"state-id\")\n@click.argument(\"state\", type=str, required=False)\n@pass_project(migrate=True)\n@click.pass_context\ndef set_state(\n    ctx: click.Context,\n    project: Project,\n    state_id: str,\n    state: str | None,\n    input_file: click.Path | None,\n    force: bool,\n):\n    \"\"\"Set state.\"\"\"\n    state_service = (\n        state_service_from_state_id(project, state_id) or ctx.obj[STATE_SERVICE_KEY]\n    )\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"set\", state_id)\n    if not reduce(xor, map(bool, [state, input_file])):\n        raise MutuallyExclusiveOptionsError(\"--input-file\", \"STATE\")\n    elif input_file:\n        with open(input_file) as state_f:\n            state_service.set_state(state_id, state_f.read())\n    elif state:\n        state_service.set_state(state_id, state)\n    logger.info(\n        f\"State for {state_id} was successfully set at {dt.utcnow():%Y-%m-%d %H:%M:%S}.\"  # noqa: WPS323\n    )\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"get\")  # noqa: WPS46\n@click.argument(\"state-id\")\n@pass_project(migrate=True)\n@click.pass_context\ndef get_state(ctx: click.Context, project: Project, state_id: str):  # noqa: WPS463\n    \"\"\"Get state.\"\"\"\n    state_service = (\n        state_service_from_state_id(project, state_id) or ctx.obj[STATE_SERVICE_KEY]\n    )\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"get\", state_id)\n    retrieved_state = state_service.get_state(state_id)\n    click.echo(json.dumps(retrieved_state))\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"clear\")\n@prompt_for_confirmation(prompt=\"This will clear state for the job. Continue?\")\n@click.argument(\"state-id\")\n@pass_project(migrate=True)\n@click.pass_context\ndef clear_state(ctx: click.Context, project: Project, state_id: str, force: bool):\n    \"\"\"Clear state.\"\"\"\n    state_service = (\n        state_service_from_state_id(project, state_id) or ctx.obj[STATE_SERVICE_KEY]\n    )\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"clear\", state_id)\n    state_service.clear_state(state_id)\n", "path": "src/meltano/cli/state.py"}], "after_files": [{"content": "\"\"\"State management in CLI.\"\"\"\nfrom __future__ import annotations\n\nimport json\nimport re\nfrom datetime import datetime as dt\nfrom functools import partial, reduce, wraps\nfrom operator import xor\n\nimport click\nimport structlog\n\nfrom meltano.cli.params import pass_project\nfrom meltano.core.block.parser import BlockParser\nfrom meltano.core.db import project_engine\nfrom meltano.core.job import Payload\nfrom meltano.core.project import Project\nfrom meltano.core.state_service import InvalidJobStateError, StateService\n\nfrom . import cli\nfrom .utils import InstrumentedCmd, InstrumentedGroup\n\nSTATE_SERVICE_KEY = \"state_service\"\n\nlogger = structlog.getLogger(__name__)\n\n\nclass MutuallyExclusiveOptionsError(Exception):\n    \"\"\"Occurs when mutually exclusive options are provided incorrectly.\"\"\"\n\n    def __init__(self, *options: str) -> None:\n        \"\"\"Instantiate the error.\n\n        Args:\n            options: the mutually exclusive options that were incorrectly provided.\n        \"\"\"\n        super().__init__(*options)\n        self.options = options\n\n    def __str__(self) -> str:\n        \"\"\"Represent the error as a string.\"\"\"\n        return f\"Must provide exactly one of: {','.join(self.options)}\"\n\n\ndef _prompt_for_confirmation(prompt):\n    \"\"\"Wrap destructive CLI commands which should prompt the user for confirmation.\"\"\"\n\n    def wrapper(func):\n        fun = click.option(\n            \"--force\", is_flag=True, help=\"Don't prompt for confirmation.\"\n        )(func)\n\n        @wraps(func)\n        def _wrapper(force=False, *args, **kwargs):\n            if force or click.confirm(prompt):\n                return fun(*args, **kwargs, force=force)\n            else:\n                click.secho(\"Aborting.\", fg=\"red\")\n\n        return _wrapper\n\n    return wrapper\n\n\nprompt_for_confirmation = partial(\n    _prompt_for_confirmation, prompt=\"This is a destructive command. Continue?\"\n)\n\n\ndef state_service_from_state_id(project: Project, state_id: str) -> StateService | None:\n    \"\"\"Instantiate by parsing a state_id.\"\"\"\n    state_id_re = re.compile(r\"^(?P<env>.+)\\:(?P<tap>.+)-to-(?P<target>.+)$\")\n    match = state_id_re.match(state_id)\n    if match:\n        # If the state_id matches convention (i.e., job has been run via \"meltano run\"),\n        # try parsing into BlockSet.\n        # This way, we get BlockSet validation and raise an error if no\n        # plugin in the BlockSet has \"state\" capability\n        try:\n            if not project.active_environment:\n                logger.warn(\n                    f\"Running state operation for environment '{match.group('env')}' outside of an environment\"\n                )\n            elif project.active_environment.name != match.group(\"env\"):\n                logger.warn(\n                    f\"Environment '{match.group('env')}' used in state operation does not match current environment '{project.active_environment.name}'.\"\n                )\n            project.activate_environment(match.group(\"env\"))\n            blocks = [match.group(\"tap\"), match.group(\"target\")]\n            parser = BlockParser(logger, project, blocks)\n            return next(parser.find_blocks()).state_service\n        except Exception:\n            logger.warn(\"No plugins found for provided state_id.\")\n    # If provided state_id does not match convention (i.e., run via \"meltano elt\"),\n    # use the standalone StateService in the CLI context.\n    return None\n\n\n@cli.group(cls=InstrumentedGroup, name=\"state\", short_help=\"Manage Singer state.\")\n@click.pass_context\n@pass_project(migrate=True)\ndef meltano_state(project: Project, ctx: click.Context):\n    \"\"\"\n    Manage state.\n\n    \\b\\nRead more at https://docs.meltano.com/reference/command-line-interface#state\n    \"\"\"\n    _, sessionmaker = project_engine(project)\n    session = sessionmaker()\n    ctx.obj[STATE_SERVICE_KEY] = StateService(session)  # noqa: WPS204\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"list\")\n@click.option(\"--pattern\", type=str, help=\"Filter state IDs by pattern.\")\n@click.pass_context\n@pass_project()\ndef list_state(\n    project: Project, ctx: click.Context, pattern: str | None\n):  # noqa: WPS125\n    \"\"\"List all state_ids for this project.\n\n    Optionally pass a glob-style pattern to filter state_ids by.\n    \"\"\"\n    state_service = ctx.obj[STATE_SERVICE_KEY]\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"list\")\n    states = state_service.list_state(pattern)\n    if states:\n        for state_id, state in states.items():\n            if state:\n                try:\n                    state_service.validate_state(json.dumps(state))\n                except (InvalidJobStateError, json.decoder.JSONDecodeError):\n                    click.secho(state_id, fg=\"red\")\n                else:\n                    click.secho(state_id, fg=\"green\")\n            else:\n                click.secho(state_id, fg=\"yellow\")\n    else:\n        logger.info(\"No state IDs found.\")\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"copy\")\n@prompt_for_confirmation(\n    prompt=\"This will overwrite state for the destination. Continue?\"\n)\n@click.argument(\"src-state-id\", type=str)\n@click.argument(\"dst-state-id\", type=str)\n@pass_project(migrate=True)\n@click.pass_context\ndef copy_state(\n    ctx: click.Context,\n    project: Project,\n    src_state_id: str,\n    dst_state_id: str,\n    force: bool,\n):\n    \"\"\"Copy state to another job id.\"\"\"\n    # Retrieve state for copying\n    state_service = (\n        state_service_from_state_id(project, src_state_id) or ctx.obj[STATE_SERVICE_KEY]\n    )\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"copy\", dst_state_id)\n\n    state_service.copy_state(src_state_id, dst_state_id)\n\n    logger.info(\n        f\"State for {dst_state_id} was successfully copied from {src_state_id} at {dt.utcnow():%Y-%m-%d %H:%M:%S}.\"  # noqa: WPS323\n    )\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"move\")\n@prompt_for_confirmation(\n    prompt=\"This will clear the source state and overwrite destination state. Continue?\"\n)\n@click.argument(\"src-state-id\", type=str)\n@click.argument(\"dst-state-id\", type=str)\n@pass_project(migrate=True)\n@click.pass_context\ndef move_state(\n    ctx: click.Context,\n    project: Project,\n    src_state_id: str,\n    dst_state_id: str,\n    force: bool,\n):\n    \"\"\"Move state to another job id, clearing the original.\"\"\"\n    # Retrieve state for moveing\n    state_service = (\n        state_service_from_state_id(project, dst_state_id) or ctx.obj[STATE_SERVICE_KEY]\n    )\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"move\", dst_state_id)\n\n    state_service.move_state(src_state_id, dst_state_id)\n\n    logger.info(\n        f\"State for {src_state_id} was successfully moved to {dst_state_id} at {dt.utcnow():%Y-%m-%d %H:%M:%S}.\"  # noqa: WPS323\n    )\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"merge\")\n@click.option(\n    \"--from-state-id\",\n    type=str,\n    help=\"Merge state from an existing state ID.\",\n)\n@click.option(\n    \"--input-file\",\n    type=click.Path(exists=True),\n    help=\"Merge state from a JSON file containing Singer state.\",\n)\n@click.argument(\"state-id\", type=str)\n@click.argument(\"state\", type=str, required=False)\n@pass_project(migrate=True)\n@click.pass_context\ndef merge_state(\n    ctx: click.Context,\n    project: Project,\n    state_id: str,\n    state: str | None,\n    input_file: click.Path | None,\n    from_state_id: str | None,\n):\n    \"\"\"Add bookmarks to existing state.\"\"\"\n    state_service = (\n        state_service_from_state_id(project, state_id) or ctx.obj[STATE_SERVICE_KEY]\n    )\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"merge\", state_id)\n    mutually_exclusive_options = [\"--input-file\", \"STATE\", \"--from-state-id\"]\n    if not reduce(xor, map(bool, [state, input_file, from_state_id])):\n        raise MutuallyExclusiveOptionsError(*mutually_exclusive_options)\n    elif input_file:\n        with open(input_file) as state_f:\n            state_service.add_state(\n                state_id, state_f.read(), payload_flags=Payload.INCOMPLETE_STATE\n            )\n    elif state:\n        state_service.add_state(state_id, state, payload_flags=Payload.INCOMPLETE_STATE)\n    elif from_state_id:\n        state_service.merge_state(from_state_id, state_id)\n    logger.info(\n        f\"State for {state_id} was successfully merged at {dt.utcnow():%Y-%m-%d %H:%M:%S}.\"  # noqa: WPS323\n    )\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"set\")\n@prompt_for_confirmation(\n    prompt=\"This will overwrite the state's current value. Continue?\"\n)\n@click.option(\n    \"--input-file\",\n    type=click.Path(exists=True),\n    help=\"Set state from json file containing Singer state.\",\n)\n@click.argument(\"state-id\")\n@click.argument(\"state\", type=str, required=False)\n@pass_project(migrate=True)\n@click.pass_context\ndef set_state(\n    ctx: click.Context,\n    project: Project,\n    state_id: str,\n    state: str | None,\n    input_file: click.Path | None,\n    force: bool,\n):\n    \"\"\"Set state.\"\"\"\n    state_service = (\n        state_service_from_state_id(project, state_id) or ctx.obj[STATE_SERVICE_KEY]\n    )\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"set\", state_id)\n    if not reduce(xor, map(bool, [state, input_file])):\n        raise MutuallyExclusiveOptionsError(\"--input-file\", \"STATE\")\n    elif input_file:\n        with open(input_file) as state_f:\n            state_service.set_state(state_id, state_f.read())\n    elif state:\n        state_service.set_state(state_id, state)\n    logger.info(\n        f\"State for {state_id} was successfully set at {dt.utcnow():%Y-%m-%d %H:%M:%S}.\"  # noqa: WPS323\n    )\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"get\")  # noqa: WPS46\n@click.argument(\"state-id\")\n@pass_project(migrate=True)\n@click.pass_context\ndef get_state(ctx: click.Context, project: Project, state_id: str):  # noqa: WPS463\n    \"\"\"Get state.\"\"\"\n    state_service = (\n        state_service_from_state_id(project, state_id) or ctx.obj[STATE_SERVICE_KEY]\n    )\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"get\", state_id)\n    retrieved_state = state_service.get_state(state_id)\n    click.echo(json.dumps(retrieved_state))\n\n\n@meltano_state.command(cls=InstrumentedCmd, name=\"clear\")\n@prompt_for_confirmation(prompt=\"This will clear state for the job. Continue?\")\n@click.argument(\"state-id\")\n@pass_project(migrate=True)\n@click.pass_context\ndef clear_state(ctx: click.Context, project: Project, state_id: str, force: bool):\n    \"\"\"Clear state.\"\"\"\n    state_service = (\n        state_service_from_state_id(project, state_id) or ctx.obj[STATE_SERVICE_KEY]\n    )\n    ctx.obj[\"legacy_tracker\"].track_meltano_state(\"clear\", state_id)\n    state_service.clear_state(state_id)\n", "path": "src/meltano/cli/state.py"}]}