We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Disallow calling a `mapper` by name in `meltano run` and alert to use a `mapping` name
The document misses a step for mapper configuration. Property `mapping_name` has to be added to the mappers extras, and its value need to match with the name under mappings. It doesn't seem been documented anywhere.

```yaml
plugins:
  mappers:
  - name: transform-field
    variant: transferwise
    pip_url: pipelinewise-transform-field
    executable: transform-field
    mapping_name: transform-field-hash
    mappings:
    - name: transform-field-hash
      config:
        transformations:
          - field_id: Name
            tap_stream_name: dbo-Test
            type: MASK-HIDDEN
```

--- 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/block/parser.py`
Content:
```
1 """Utilities for turning a string list of plugins into a usable list of BlockSet and PluginCommand objects."""
2 from typing import Dict, Generator, List, Optional, Tuple, Union
3 
4 import click
5 import structlog
6 
7 from meltano.core.plugin import PluginType
8 from meltano.core.plugin.error import PluginNotFoundError
9 from meltano.core.plugin.project_plugin import ProjectPlugin
10 from meltano.core.project_plugins_service import ProjectPluginsService
11 from meltano.core.task_sets_service import TaskSetsService
12 
13 from .blockset import BlockSet, BlockSetValidationError
14 from .extract_load import ELBContextBuilder, ExtractLoadBlocks
15 from .plugin_command import PluginCommandBlock, plugin_command_invoker
16 from .singer import CONSUMERS, SingerBlock
17 
18 
19 def is_command_block(plugin: ProjectPlugin) -> bool:
20     """Check if a plugin is a command block.
21 
22     Args:
23         plugin: Plugin to check.
24 
25     Returns:
26         True if plugin is a command block.
27     """
28     return plugin.type not in {
29         PluginType.EXTRACTORS,
30         PluginType.LOADERS,
31         PluginType.MAPPERS,
32     }
33 
34 
35 def validate_block_sets(
36     log: structlog.BoundLogger, blocks: List[Union[BlockSet, PluginCommandBlock]]
37 ) -> bool:
38     """Perform validation of all blocks in a list that implement the BlockSet interface.
39 
40     Args:
41         log: Logger to use in the event of a validation error.
42         blocks: A list of blocks.
43 
44     Returns:
45         True if all blocks are valid, False otherwise.
46     """
47     for idx, blk in enumerate(blocks):
48         if blk == BlockSet:
49             log.debug("validating ExtractLoadBlock.", set_number=idx)
50             try:
51                 blk.validate_set()
52             except Exception as err:
53                 log.error("Validation failed.", err=err)
54                 return False
55     return True
56 
57 
58 class BlockParser:  # noqa: D101
59     def __init__(
60         self,
61         log: structlog.BoundLogger,
62         project,
63         blocks: List[str],
64         full_refresh: Optional[bool] = False,
65         no_state_update: Optional[bool] = False,
66         force: Optional[bool] = False,
67     ):
68         """
69         Parse a meltano run command invocation into a list of blocks.
70 
71         Args:
72             log: Logger to use.
73             project: Project to use.
74             blocks: List of block names to parse.
75             full_refresh: Whether to perform a full refresh (applies to all found sets).
76             no_state_update: Whether to run with or without state updates.
77             force: Whether to force a run if a job is already running (applies to all found sets).
78 
79         Raises:
80             ClickException: If a block name is not found.
81         """
82         self.log = log
83         self.project = project
84 
85         self._full_refresh = full_refresh
86         self._no_state_update = no_state_update
87         self._force = force
88 
89         self._plugins_service = ProjectPluginsService(project)
90         self._plugins: List[ProjectPlugin] = []
91 
92         self._commands: Dict[int, str] = {}
93         self._mappings_ref: Dict[int, str] = {}
94 
95         task_sets_service: TaskSetsService = TaskSetsService(project)
96 
97         blocks = self._expand_jobs(blocks, task_sets_service)
98 
99         for idx, name in enumerate(blocks):
100 
101             try:
102                 parsed_name, command_name = name.split(":")
103             except ValueError:
104                 parsed_name = name
105                 command_name = None
106 
107             plugin = self._find_plugin_or_mapping(parsed_name)
108             if plugin is None:
109                 raise click.ClickException(f"Block {name} not found")
110 
111             if plugin and task_sets_service.exists(name):
112                 raise click.ClickException(
113                     f"Ambiguous reference to '{name}' which matches a job name AND a plugin name."
114                 )
115 
116             if plugin.type == PluginType.MAPPERS:
117                 self._mappings_ref[idx] = parsed_name
118 
119             self._plugins.append(plugin)
120             if command_name:
121                 self._commands[idx] = command_name
122                 self.log.debug(
123                     "plugin command added for execution",
124                     commands=self._commands,
125                     command_name=command_name,
126                     plugin_name=parsed_name,
127                 )
128 
129             self.log.debug("found plugin in cli invocation", plugin_name=plugin.name)
130 
131     def _expand_jobs(self, blocks: List[str], task_sets: TaskSetsService) -> List[str]:
132         """Expand any jobs present in a list of blocks into their raw block names.
133 
134         Example:
135             Given a job named "somejob" which consists of a single task of "tap target":
136             ["somejob", "dbt:run"] -> ["tap", "target", "dbt:run"]
137 
138         Args:
139             blocks: List of block names to parse.
140             task_sets: TaskSetsService to use.
141 
142         Returns:
143             List of block names with jobs expanded.
144         """
145         expanded_blocks: List[str] = []
146         for name in blocks:
147             if task_sets.exists(name):
148                 self.log.debug(
149                     "expanding job to tasks",
150                     job_name=name,
151                     tasks=task_sets.get(name).flat_args,
152                 )
153                 expanded_blocks.extend(task_sets.get(name).flat_args)
154             else:
155                 expanded_blocks.append(name)
156         return expanded_blocks
157 
158     def find_blocks(
159         self, offset: int = 0
160     ) -> Generator[Union[BlockSet, PluginCommandBlock], None, None]:
161         """
162         Find all blocks in the invocation.
163 
164         Args:
165             offset: Offset to start from.
166 
167         Yields:
168             Generator of blocks (either BlockSet or PluginCommandBlock).
169 
170         Raises:
171             BlockSetValidationError: If unknown command is found or if a unexpected block sequence is found.
172         """
173         cur = offset
174         while cur < len(self._plugins):
175             plugin = self._plugins[cur]
176             elb, idx = self._find_next_elb_set(cur)
177             if elb:
178                 self.log.debug("found ExtractLoadBlocks set", offset=cur)
179                 yield elb
180                 cur += idx
181             elif is_command_block(plugin):
182                 self.log.debug(
183                     "found PluginCommand",
184                     offset=cur,
185                     plugin_type=plugin.type,
186                 )
187                 yield plugin_command_invoker(
188                     self._plugins[cur],
189                     self.project,
190                     command=self._commands.get(cur),
191                 )
192                 cur += 1
193             else:
194                 raise BlockSetValidationError(
195                     f"Unknown command type or bad block sequence at index {cur + 1}, starting block '{plugin.name}'"  # noqa: WPS237
196                 )
197 
198     def _find_plugin_or_mapping(self, name: str) -> Optional[ProjectPlugin]:
199         """Find a plugin by name OR by mapping name.
200 
201         Args:
202             name: Name of the plugin or mapping.
203 
204         Returns:
205             The actual plugin.
206 
207         Raises:
208             ClickException: If mapping name returns multiple matches.
209         """
210         try:
211             return self._plugins_service.find_plugin(name)
212         except PluginNotFoundError:
213             pass
214 
215         mapper = None
216         try:
217             mapper = self._plugins_service.find_plugins_by_mapping_name(name)
218         except PluginNotFoundError:
219             pass
220 
221         if mapper is None:
222             return None
223 
224         if len(mapper) > 1:
225             raise click.ClickException(
226                 f"Ambiguous mapping name {name}, found multiple matches."
227             )
228         return mapper[0] if mapper else None
229 
230     def _find_next_elb_set(  # noqa: WPS231, WPS213
231         self,
232         offset: int = 0,
233     ) -> Tuple[Optional[ExtractLoadBlocks], int]:  # noqa: WPS231, WPS213
234         """
235         Search a list of project plugins trying to find an extract ExtractLoad block set.
236 
237         Args:
238             offset: Optional starting offset for search.
239 
240         Returns:
241             The ExtractLoad object.
242             Offset for remaining plugins.
243 
244         Raises:
245             BlockSetValidationError: If the block set is not valid.
246         """
247         blocks: List[SingerBlock] = []
248 
249         base_builder = ELBContextBuilder(
250             self.project, self._plugins_service
251         )  # lint work around
252         builder = (
253             base_builder.with_force(self._force)
254             .with_full_refresh(self._full_refresh)
255             .with_no_state_update(self._no_state_update)
256         )
257 
258         if self._plugins[offset].type != PluginType.EXTRACTORS:
259             self.log.debug(
260                 "next block not extractor",
261                 offset=offset,
262                 plugin_type=self._plugins[offset].type,
263             )
264             return None, offset
265 
266         self.log.debug(
267             "head of set is extractor as expected", block=self._plugins[offset]
268         )
269 
270         blocks.append(builder.make_block(self._plugins[offset]))
271 
272         for idx, plugin in enumerate(self._plugins[offset + 1 :]):  # noqa: E203
273             next_block = idx + 1
274 
275             if plugin.type not in CONSUMERS:
276                 self.log.debug(
277                     "next block not a consumer of output",
278                     offset=offset,
279                     plugin_type=plugin.type,
280                 )
281                 return None, offset + next_block
282 
283             self.log.debug("found block", block_type=plugin.type, index=next_block)
284 
285             if plugin.type == PluginType.MAPPERS:
286                 self.log.debug(
287                     "found mapper",
288                     plugin_type=plugin.type,
289                     plugin_name=plugin.name,
290                     mapping=self._mappings_ref.get(next_block),
291                     idx=next_block,
292                 )
293                 blocks.append(
294                     builder.make_block(
295                         plugin,
296                     )
297                 )
298             elif plugin.type == PluginType.LOADERS:
299                 self.log.debug("blocks", offset=offset, idx=next_block)
300                 blocks.append(builder.make_block(plugin))
301                 elb = ExtractLoadBlocks(builder.context(), blocks)
302                 return elb, idx + 2
303             else:
304                 self.log.warning(
305                     "Found unexpected plugin type for block in middle of block set.",
306                     plugin_type=plugin.type,
307                     plugin_name=plugin.name,
308                 )
309                 raise BlockSetValidationError(
310                     f"Expected {PluginType.MAPPERS} or {PluginType.LOADERS}."
311                 )
312         raise BlockSetValidationError("Found no end in block set!")
313
```
--- 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/core/block/parser.py b/src/meltano/core/block/parser.py
--- a/src/meltano/core/block/parser.py
+++ b/src/meltano/core/block/parser.py
@@ -290,11 +290,25 @@
                     mapping=self._mappings_ref.get(next_block),
                     idx=next_block,
                 )
-                blocks.append(
-                    builder.make_block(
-                        plugin,
+                # Checks to see if the mapper plugin name is the same as the mappings name
+                # If they both match then a validation error is raised because the
+                # meltano run command needs the mappings name to obtain the settings to
+                # pass to the parent mapper plugin.  We also want to fail if the user names them
+                # the same to stop errors due to ambiguous commands.
+                if plugin.name == self._mappings_ref.get(next_block):
+                    self.log.warning(
+                        "Found unexpected mapper plugin name. ",
+                        plugin_name=plugin.name,
+                    )
+                    raise BlockSetValidationError(
+                        f"Expected unique mappings name not the mapper plugin name: {plugin.name}."
+                    )
+                else:
+                    blocks.append(
+                        builder.make_block(
+                            plugin,
+                        )
                     )
-                )
             elif plugin.type == PluginType.LOADERS:
                 self.log.debug("blocks", offset=offset, idx=next_block)
                 blocks.append(builder.make_block(plugin))

{"golden_diff": "diff --git a/src/meltano/core/block/parser.py b/src/meltano/core/block/parser.py\n--- a/src/meltano/core/block/parser.py\n+++ b/src/meltano/core/block/parser.py\n@@ -290,11 +290,25 @@\n                     mapping=self._mappings_ref.get(next_block),\n                     idx=next_block,\n                 )\n-                blocks.append(\n-                    builder.make_block(\n-                        plugin,\n+                # Checks to see if the mapper plugin name is the same as the mappings name\n+                # If they both match then a validation error is raised because the\n+                # meltano run command needs the mappings name to obtain the settings to\n+                # pass to the parent mapper plugin.  We also want to fail if the user names them\n+                # the same to stop errors due to ambiguous commands.\n+                if plugin.name == self._mappings_ref.get(next_block):\n+                    self.log.warning(\n+                        \"Found unexpected mapper plugin name. \",\n+                        plugin_name=plugin.name,\n+                    )\n+                    raise BlockSetValidationError(\n+                        f\"Expected unique mappings name not the mapper plugin name: {plugin.name}.\"\n+                    )\n+                else:\n+                    blocks.append(\n+                        builder.make_block(\n+                            plugin,\n+                        )\n                     )\n-                )\n             elif plugin.type == PluginType.LOADERS:\n                 self.log.debug(\"blocks\", offset=offset, idx=next_block)\n                 blocks.append(builder.make_block(plugin))\n", "issue": "Disallow calling a `mapper` by name in `meltano run` and alert to use a `mapping` name\nThe document misses a step for mapper configuration. Property `mapping_name` has to be added to the mappers extras, and its value need to match with the name under mappings. It doesn't seem been documented anywhere.\r\n\r\n```yaml\r\nplugins:\r\n  mappers:\r\n  - name: transform-field\r\n    variant: transferwise\r\n    pip_url: pipelinewise-transform-field\r\n    executable: transform-field\r\n    mapping_name: transform-field-hash\r\n    mappings:\r\n    - name: transform-field-hash\r\n      config:\r\n        transformations:\r\n          - field_id: Name\r\n            tap_stream_name: dbo-Test\r\n            type: MASK-HIDDEN\r\n```\n", "before_files": [{"content": "\"\"\"Utilities for turning a string list of plugins into a usable list of BlockSet and PluginCommand objects.\"\"\"\nfrom typing import Dict, Generator, List, Optional, Tuple, Union\n\nimport click\nimport structlog\n\nfrom meltano.core.plugin import PluginType\nfrom meltano.core.plugin.error import PluginNotFoundError\nfrom meltano.core.plugin.project_plugin import ProjectPlugin\nfrom meltano.core.project_plugins_service import ProjectPluginsService\nfrom meltano.core.task_sets_service import TaskSetsService\n\nfrom .blockset import BlockSet, BlockSetValidationError\nfrom .extract_load import ELBContextBuilder, ExtractLoadBlocks\nfrom .plugin_command import PluginCommandBlock, plugin_command_invoker\nfrom .singer import CONSUMERS, SingerBlock\n\n\ndef is_command_block(plugin: ProjectPlugin) -> bool:\n    \"\"\"Check if a plugin is a command block.\n\n    Args:\n        plugin: Plugin to check.\n\n    Returns:\n        True if plugin is a command block.\n    \"\"\"\n    return plugin.type not in {\n        PluginType.EXTRACTORS,\n        PluginType.LOADERS,\n        PluginType.MAPPERS,\n    }\n\n\ndef validate_block_sets(\n    log: structlog.BoundLogger, blocks: List[Union[BlockSet, PluginCommandBlock]]\n) -> bool:\n    \"\"\"Perform validation of all blocks in a list that implement the BlockSet interface.\n\n    Args:\n        log: Logger to use in the event of a validation error.\n        blocks: A list of blocks.\n\n    Returns:\n        True if all blocks are valid, False otherwise.\n    \"\"\"\n    for idx, blk in enumerate(blocks):\n        if blk == BlockSet:\n            log.debug(\"validating ExtractLoadBlock.\", set_number=idx)\n            try:\n                blk.validate_set()\n            except Exception as err:\n                log.error(\"Validation failed.\", err=err)\n                return False\n    return True\n\n\nclass BlockParser:  # noqa: D101\n    def __init__(\n        self,\n        log: structlog.BoundLogger,\n        project,\n        blocks: List[str],\n        full_refresh: Optional[bool] = False,\n        no_state_update: Optional[bool] = False,\n        force: Optional[bool] = False,\n    ):\n        \"\"\"\n        Parse a meltano run command invocation into a list of blocks.\n\n        Args:\n            log: Logger to use.\n            project: Project to use.\n            blocks: List of block names to parse.\n            full_refresh: Whether to perform a full refresh (applies to all found sets).\n            no_state_update: Whether to run with or without state updates.\n            force: Whether to force a run if a job is already running (applies to all found sets).\n\n        Raises:\n            ClickException: If a block name is not found.\n        \"\"\"\n        self.log = log\n        self.project = project\n\n        self._full_refresh = full_refresh\n        self._no_state_update = no_state_update\n        self._force = force\n\n        self._plugins_service = ProjectPluginsService(project)\n        self._plugins: List[ProjectPlugin] = []\n\n        self._commands: Dict[int, str] = {}\n        self._mappings_ref: Dict[int, str] = {}\n\n        task_sets_service: TaskSetsService = TaskSetsService(project)\n\n        blocks = self._expand_jobs(blocks, task_sets_service)\n\n        for idx, name in enumerate(blocks):\n\n            try:\n                parsed_name, command_name = name.split(\":\")\n            except ValueError:\n                parsed_name = name\n                command_name = None\n\n            plugin = self._find_plugin_or_mapping(parsed_name)\n            if plugin is None:\n                raise click.ClickException(f\"Block {name} not found\")\n\n            if plugin and task_sets_service.exists(name):\n                raise click.ClickException(\n                    f\"Ambiguous reference to '{name}' which matches a job name AND a plugin name.\"\n                )\n\n            if plugin.type == PluginType.MAPPERS:\n                self._mappings_ref[idx] = parsed_name\n\n            self._plugins.append(plugin)\n            if command_name:\n                self._commands[idx] = command_name\n                self.log.debug(\n                    \"plugin command added for execution\",\n                    commands=self._commands,\n                    command_name=command_name,\n                    plugin_name=parsed_name,\n                )\n\n            self.log.debug(\"found plugin in cli invocation\", plugin_name=plugin.name)\n\n    def _expand_jobs(self, blocks: List[str], task_sets: TaskSetsService) -> List[str]:\n        \"\"\"Expand any jobs present in a list of blocks into their raw block names.\n\n        Example:\n            Given a job named \"somejob\" which consists of a single task of \"tap target\":\n            [\"somejob\", \"dbt:run\"] -> [\"tap\", \"target\", \"dbt:run\"]\n\n        Args:\n            blocks: List of block names to parse.\n            task_sets: TaskSetsService to use.\n\n        Returns:\n            List of block names with jobs expanded.\n        \"\"\"\n        expanded_blocks: List[str] = []\n        for name in blocks:\n            if task_sets.exists(name):\n                self.log.debug(\n                    \"expanding job to tasks\",\n                    job_name=name,\n                    tasks=task_sets.get(name).flat_args,\n                )\n                expanded_blocks.extend(task_sets.get(name).flat_args)\n            else:\n                expanded_blocks.append(name)\n        return expanded_blocks\n\n    def find_blocks(\n        self, offset: int = 0\n    ) -> Generator[Union[BlockSet, PluginCommandBlock], None, None]:\n        \"\"\"\n        Find all blocks in the invocation.\n\n        Args:\n            offset: Offset to start from.\n\n        Yields:\n            Generator of blocks (either BlockSet or PluginCommandBlock).\n\n        Raises:\n            BlockSetValidationError: If unknown command is found or if a unexpected block sequence is found.\n        \"\"\"\n        cur = offset\n        while cur < len(self._plugins):\n            plugin = self._plugins[cur]\n            elb, idx = self._find_next_elb_set(cur)\n            if elb:\n                self.log.debug(\"found ExtractLoadBlocks set\", offset=cur)\n                yield elb\n                cur += idx\n            elif is_command_block(plugin):\n                self.log.debug(\n                    \"found PluginCommand\",\n                    offset=cur,\n                    plugin_type=plugin.type,\n                )\n                yield plugin_command_invoker(\n                    self._plugins[cur],\n                    self.project,\n                    command=self._commands.get(cur),\n                )\n                cur += 1\n            else:\n                raise BlockSetValidationError(\n                    f\"Unknown command type or bad block sequence at index {cur + 1}, starting block '{plugin.name}'\"  # noqa: WPS237\n                )\n\n    def _find_plugin_or_mapping(self, name: str) -> Optional[ProjectPlugin]:\n        \"\"\"Find a plugin by name OR by mapping name.\n\n        Args:\n            name: Name of the plugin or mapping.\n\n        Returns:\n            The actual plugin.\n\n        Raises:\n            ClickException: If mapping name returns multiple matches.\n        \"\"\"\n        try:\n            return self._plugins_service.find_plugin(name)\n        except PluginNotFoundError:\n            pass\n\n        mapper = None\n        try:\n            mapper = self._plugins_service.find_plugins_by_mapping_name(name)\n        except PluginNotFoundError:\n            pass\n\n        if mapper is None:\n            return None\n\n        if len(mapper) > 1:\n            raise click.ClickException(\n                f\"Ambiguous mapping name {name}, found multiple matches.\"\n            )\n        return mapper[0] if mapper else None\n\n    def _find_next_elb_set(  # noqa: WPS231, WPS213\n        self,\n        offset: int = 0,\n    ) -> Tuple[Optional[ExtractLoadBlocks], int]:  # noqa: WPS231, WPS213\n        \"\"\"\n        Search a list of project plugins trying to find an extract ExtractLoad block set.\n\n        Args:\n            offset: Optional starting offset for search.\n\n        Returns:\n            The ExtractLoad object.\n            Offset for remaining plugins.\n\n        Raises:\n            BlockSetValidationError: If the block set is not valid.\n        \"\"\"\n        blocks: List[SingerBlock] = []\n\n        base_builder = ELBContextBuilder(\n            self.project, self._plugins_service\n        )  # lint work around\n        builder = (\n            base_builder.with_force(self._force)\n            .with_full_refresh(self._full_refresh)\n            .with_no_state_update(self._no_state_update)\n        )\n\n        if self._plugins[offset].type != PluginType.EXTRACTORS:\n            self.log.debug(\n                \"next block not extractor\",\n                offset=offset,\n                plugin_type=self._plugins[offset].type,\n            )\n            return None, offset\n\n        self.log.debug(\n            \"head of set is extractor as expected\", block=self._plugins[offset]\n        )\n\n        blocks.append(builder.make_block(self._plugins[offset]))\n\n        for idx, plugin in enumerate(self._plugins[offset + 1 :]):  # noqa: E203\n            next_block = idx + 1\n\n            if plugin.type not in CONSUMERS:\n                self.log.debug(\n                    \"next block not a consumer of output\",\n                    offset=offset,\n                    plugin_type=plugin.type,\n                )\n                return None, offset + next_block\n\n            self.log.debug(\"found block\", block_type=plugin.type, index=next_block)\n\n            if plugin.type == PluginType.MAPPERS:\n                self.log.debug(\n                    \"found mapper\",\n                    plugin_type=plugin.type,\n                    plugin_name=plugin.name,\n                    mapping=self._mappings_ref.get(next_block),\n                    idx=next_block,\n                )\n                blocks.append(\n                    builder.make_block(\n                        plugin,\n                    )\n                )\n            elif plugin.type == PluginType.LOADERS:\n                self.log.debug(\"blocks\", offset=offset, idx=next_block)\n                blocks.append(builder.make_block(plugin))\n                elb = ExtractLoadBlocks(builder.context(), blocks)\n                return elb, idx + 2\n            else:\n                self.log.warning(\n                    \"Found unexpected plugin type for block in middle of block set.\",\n                    plugin_type=plugin.type,\n                    plugin_name=plugin.name,\n                )\n                raise BlockSetValidationError(\n                    f\"Expected {PluginType.MAPPERS} or {PluginType.LOADERS}.\"\n                )\n        raise BlockSetValidationError(\"Found no end in block set!\")\n", "path": "src/meltano/core/block/parser.py"}], "after_files": [{"content": "\"\"\"Utilities for turning a string list of plugins into a usable list of BlockSet and PluginCommand objects.\"\"\"\nfrom typing import Dict, Generator, List, Optional, Tuple, Union\n\nimport click\nimport structlog\n\nfrom meltano.core.plugin import PluginType\nfrom meltano.core.plugin.error import PluginNotFoundError\nfrom meltano.core.plugin.project_plugin import ProjectPlugin\nfrom meltano.core.project_plugins_service import ProjectPluginsService\nfrom meltano.core.task_sets_service import TaskSetsService\n\nfrom .blockset import BlockSet, BlockSetValidationError\nfrom .extract_load import ELBContextBuilder, ExtractLoadBlocks\nfrom .plugin_command import PluginCommandBlock, plugin_command_invoker\nfrom .singer import CONSUMERS, SingerBlock\n\n\ndef is_command_block(plugin: ProjectPlugin) -> bool:\n    \"\"\"Check if a plugin is a command block.\n\n    Args:\n        plugin: Plugin to check.\n\n    Returns:\n        True if plugin is a command block.\n    \"\"\"\n    return plugin.type not in {\n        PluginType.EXTRACTORS,\n        PluginType.LOADERS,\n        PluginType.MAPPERS,\n    }\n\n\ndef validate_block_sets(\n    log: structlog.BoundLogger, blocks: List[Union[BlockSet, PluginCommandBlock]]\n) -> bool:\n    \"\"\"Perform validation of all blocks in a list that implement the BlockSet interface.\n\n    Args:\n        log: Logger to use in the event of a validation error.\n        blocks: A list of blocks.\n\n    Returns:\n        True if all blocks are valid, False otherwise.\n    \"\"\"\n    for idx, blk in enumerate(blocks):\n        if blk == BlockSet:\n            log.debug(\"validating ExtractLoadBlock.\", set_number=idx)\n            try:\n                blk.validate_set()\n            except Exception as err:\n                log.error(\"Validation failed.\", err=err)\n                return False\n    return True\n\n\nclass BlockParser:  # noqa: D101\n    def __init__(\n        self,\n        log: structlog.BoundLogger,\n        project,\n        blocks: List[str],\n        full_refresh: Optional[bool] = False,\n        no_state_update: Optional[bool] = False,\n        force: Optional[bool] = False,\n    ):\n        \"\"\"\n        Parse a meltano run command invocation into a list of blocks.\n\n        Args:\n            log: Logger to use.\n            project: Project to use.\n            blocks: List of block names to parse.\n            full_refresh: Whether to perform a full refresh (applies to all found sets).\n            no_state_update: Whether to run with or without state updates.\n            force: Whether to force a run if a job is already running (applies to all found sets).\n\n        Raises:\n            ClickException: If a block name is not found.\n        \"\"\"\n        self.log = log\n        self.project = project\n\n        self._full_refresh = full_refresh\n        self._no_state_update = no_state_update\n        self._force = force\n\n        self._plugins_service = ProjectPluginsService(project)\n        self._plugins: List[ProjectPlugin] = []\n\n        self._commands: Dict[int, str] = {}\n        self._mappings_ref: Dict[int, str] = {}\n\n        task_sets_service: TaskSetsService = TaskSetsService(project)\n\n        blocks = self._expand_jobs(blocks, task_sets_service)\n\n        for idx, name in enumerate(blocks):\n\n            try:\n                parsed_name, command_name = name.split(\":\")\n            except ValueError:\n                parsed_name = name\n                command_name = None\n\n            plugin = self._find_plugin_or_mapping(parsed_name)\n            if plugin is None:\n                raise click.ClickException(f\"Block {name} not found\")\n\n            if plugin and task_sets_service.exists(name):\n                raise click.ClickException(\n                    f\"Ambiguous reference to '{name}' which matches a job name AND a plugin name.\"\n                )\n\n            if plugin.type == PluginType.MAPPERS:\n                self._mappings_ref[idx] = parsed_name\n\n            self._plugins.append(plugin)\n            if command_name:\n                self._commands[idx] = command_name\n                self.log.debug(\n                    \"plugin command added for execution\",\n                    commands=self._commands,\n                    command_name=command_name,\n                    plugin_name=parsed_name,\n                )\n\n            self.log.debug(\"found plugin in cli invocation\", plugin_name=plugin.name)\n\n    def _expand_jobs(self, blocks: List[str], task_sets: TaskSetsService) -> List[str]:\n        \"\"\"Expand any jobs present in a list of blocks into their raw block names.\n\n        Example:\n            Given a job named \"somejob\" which consists of a single task of \"tap target\":\n            [\"somejob\", \"dbt:run\"] -> [\"tap\", \"target\", \"dbt:run\"]\n\n        Args:\n            blocks: List of block names to parse.\n            task_sets: TaskSetsService to use.\n\n        Returns:\n            List of block names with jobs expanded.\n        \"\"\"\n        expanded_blocks: List[str] = []\n        for name in blocks:\n            if task_sets.exists(name):\n                self.log.debug(\n                    \"expanding job to tasks\",\n                    job_name=name,\n                    tasks=task_sets.get(name).flat_args,\n                )\n                expanded_blocks.extend(task_sets.get(name).flat_args)\n            else:\n                expanded_blocks.append(name)\n        return expanded_blocks\n\n    def find_blocks(\n        self, offset: int = 0\n    ) -> Generator[Union[BlockSet, PluginCommandBlock], None, None]:\n        \"\"\"\n        Find all blocks in the invocation.\n\n        Args:\n            offset: Offset to start from.\n\n        Yields:\n            Generator of blocks (either BlockSet or PluginCommandBlock).\n\n        Raises:\n            BlockSetValidationError: If unknown command is found or if a unexpected block sequence is found.\n        \"\"\"\n        cur = offset\n        while cur < len(self._plugins):\n            plugin = self._plugins[cur]\n            elb, idx = self._find_next_elb_set(cur)\n            if elb:\n                self.log.debug(\"found ExtractLoadBlocks set\", offset=cur)\n                yield elb\n                cur += idx\n            elif is_command_block(plugin):\n                self.log.debug(\n                    \"found PluginCommand\",\n                    offset=cur,\n                    plugin_type=plugin.type,\n                )\n                yield plugin_command_invoker(\n                    self._plugins[cur],\n                    self.project,\n                    command=self._commands.get(cur),\n                )\n                cur += 1\n            else:\n                raise BlockSetValidationError(\n                    f\"Unknown command type or bad block sequence at index {cur + 1}, starting block '{plugin.name}'\"  # noqa: WPS237\n                )\n\n    def _find_plugin_or_mapping(self, name: str) -> Optional[ProjectPlugin]:\n        \"\"\"Find a plugin by name OR by mapping name.\n\n        Args:\n            name: Name of the plugin or mapping.\n\n        Returns:\n            The actual plugin.\n\n        Raises:\n            ClickException: If mapping name returns multiple matches.\n        \"\"\"\n        try:\n            return self._plugins_service.find_plugin(name)\n        except PluginNotFoundError:\n            pass\n\n        mapper = None\n        try:\n            mapper = self._plugins_service.find_plugins_by_mapping_name(name)\n        except PluginNotFoundError:\n            pass\n\n        if mapper is None:\n            return None\n\n        if len(mapper) > 1:\n            raise click.ClickException(\n                f\"Ambiguous mapping name {name}, found multiple matches.\"\n            )\n        return mapper[0] if mapper else None\n\n    def _find_next_elb_set(  # noqa: WPS231, WPS213\n        self,\n        offset: int = 0,\n    ) -> Tuple[Optional[ExtractLoadBlocks], int]:  # noqa: WPS231, WPS213\n        \"\"\"\n        Search a list of project plugins trying to find an extract ExtractLoad block set.\n\n        Args:\n            offset: Optional starting offset for search.\n\n        Returns:\n            The ExtractLoad object.\n            Offset for remaining plugins.\n\n        Raises:\n            BlockSetValidationError: If the block set is not valid.\n        \"\"\"\n        blocks: List[SingerBlock] = []\n\n        base_builder = ELBContextBuilder(\n            self.project, self._plugins_service\n        )  # lint work around\n        builder = (\n            base_builder.with_force(self._force)\n            .with_full_refresh(self._full_refresh)\n            .with_no_state_update(self._no_state_update)\n        )\n\n        if self._plugins[offset].type != PluginType.EXTRACTORS:\n            self.log.debug(\n                \"next block not extractor\",\n                offset=offset,\n                plugin_type=self._plugins[offset].type,\n            )\n            return None, offset\n\n        self.log.debug(\n            \"head of set is extractor as expected\", block=self._plugins[offset]\n        )\n\n        blocks.append(builder.make_block(self._plugins[offset]))\n\n        for idx, plugin in enumerate(self._plugins[offset + 1 :]):  # noqa: E203\n            next_block = idx + 1\n\n            if plugin.type not in CONSUMERS:\n                self.log.debug(\n                    \"next block not a consumer of output\",\n                    offset=offset,\n                    plugin_type=plugin.type,\n                )\n                return None, offset + next_block\n\n            self.log.debug(\"found block\", block_type=plugin.type, index=next_block)\n\n            if plugin.type == PluginType.MAPPERS:\n                self.log.debug(\n                    \"found mapper\",\n                    plugin_type=plugin.type,\n                    plugin_name=plugin.name,\n                    mapping=self._mappings_ref.get(next_block),\n                    idx=next_block,\n                )\n                # Checks to see if the mapper plugin name is the same as the mappings name\n                # If they both match then a validation error is raised because the\n                # meltano run command needs the mappings name to obtain the settings to\n                # pass to the parent mapper plugin.  We also want to fail if the user names them\n                # the same to stop errors due to ambiguous commands.\n                if plugin.name == self._mappings_ref.get(next_block):\n                    self.log.warning(\n                        \"Found unexpected mapper plugin name. \",\n                        plugin_name=plugin.name,\n                    )\n                    raise BlockSetValidationError(\n                        f\"Expected unique mappings name not the mapper plugin name: {plugin.name}.\"\n                    )\n                else:\n                    blocks.append(\n                        builder.make_block(\n                            plugin,\n                        )\n                    )\n            elif plugin.type == PluginType.LOADERS:\n                self.log.debug(\"blocks\", offset=offset, idx=next_block)\n                blocks.append(builder.make_block(plugin))\n                elb = ExtractLoadBlocks(builder.context(), blocks)\n                return elb, idx + 2\n            else:\n                self.log.warning(\n                    \"Found unexpected plugin type for block in middle of block set.\",\n                    plugin_type=plugin.type,\n                    plugin_name=plugin.name,\n                )\n                raise BlockSetValidationError(\n                    f\"Expected {PluginType.MAPPERS} or {PluginType.LOADERS}.\"\n                )\n        raise BlockSetValidationError(\"Found no end in block set!\")\n", "path": "src/meltano/core/block/parser.py"}]}