We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Final clean up for `kedro pipeline create` and `kedro catalog create` docs
## Description
<!-- Is your feature request related to a problem? A clear and concise description of what the problem is: "I'm always frustrated when ..." -->
Steps to reprdouce:
```
kedro new -s spaceflights --checkout main 
kedro pipeline create new
```

## Inconsistency 1 - `kedro pipeline create` add `README.md` while starter doesn't have it
![image](https://github.com/kedro-org/kedro/assets/18221871/f8b75c44-4613-4769-8f06-5a75f0e633c7)

## Inconsistency 2 - CLI docstring seems to be outdated.

https://github.com/kedro-org/kedro/blob/6888001c6019059ae717c99fbe26a06f67d73ca2/kedro/framework/cli/catalog.py#L131C1-L141C8

## Inconsistency 3 - broken doc
It generates a link that doesn't exist - https://docs.kedro.org/en/0.18.12/kedro_project_setup/configuration.html#parameters

See this:
https://github.com/kedro-org/kedro/blob/6888001c6019059ae717c99fbe26a06f67d73ca2/kedro/templates/pipeline/%7B%7B%20cookiecutter.pipeline_name%20%7D%7D/config/parameters_%7B%7B%20cookiecutter.pipeline_name%20%7D%7D.yml#L5

Also, any idea it is in a folder called `config` but not `conf`?
## Outdated Doc 4
See https://github.com/kedro-org/kedro/pull/2888#discussion_r1297170881



## Context
<!-- Why is this change important to you? How would you use it? How can it benefit other users? -->

## Possible Implementation
<!-- (Optional) Suggest an idea for implementing the addition or change. -->

## Possible Alternatives
<!-- (Optional) Describe any alternative solutions or features you've considered. -->


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `kedro/framework/cli/catalog.py`
Content:
```
1 """A collection of CLI commands for working with Kedro catalog."""
2 from collections import defaultdict
3 from itertools import chain
4 
5 import click
6 import yaml
7 from click import secho
8 
9 from kedro.framework.cli.utils import KedroCliError, env_option, split_string
10 from kedro.framework.project import pipelines, settings
11 from kedro.framework.session import KedroSession
12 from kedro.framework.startup import ProjectMetadata
13 
14 
15 def _create_session(package_name: str, **kwargs):
16     kwargs.setdefault("save_on_close", False)
17     try:
18         return KedroSession.create(package_name, **kwargs)
19     except Exception as exc:
20         raise KedroCliError(
21             f"Unable to instantiate Kedro session.\nError: {exc}"
22         ) from exc
23 
24 
25 # noqa: missing-function-docstring
26 @click.group(name="Kedro")
27 def catalog_cli():  # pragma: no cover
28     pass
29 
30 
31 @catalog_cli.group()
32 def catalog():
33     """Commands for working with catalog."""
34 
35 
36 # noqa: too-many-locals,protected-access
37 @catalog.command("list")
38 @env_option
39 @click.option(
40     "--pipeline",
41     "-p",
42     type=str,
43     default="",
44     help="Name of the modular pipeline to run. If not set, "
45     "the project pipeline is run by default.",
46     callback=split_string,
47 )
48 @click.pass_obj
49 def list_datasets(metadata: ProjectMetadata, pipeline, env):
50     """Show datasets per type."""
51     title = "Datasets in '{}' pipeline"
52     not_mentioned = "Datasets not mentioned in pipeline"
53     mentioned = "Datasets mentioned in pipeline"
54     factories = "Datasets generated from factories"
55 
56     session = _create_session(metadata.package_name, env=env)
57     context = session.load_context()
58 
59     data_catalog = context.catalog
60     datasets_meta = data_catalog._data_sets
61     catalog_ds = set(data_catalog.list())
62 
63     target_pipelines = pipeline or pipelines.keys()
64 
65     result = {}
66     for pipe in target_pipelines:
67         pl_obj = pipelines.get(pipe)
68         if pl_obj:
69             pipeline_ds = pl_obj.data_sets()
70         else:
71             existing_pls = ", ".join(sorted(pipelines.keys()))
72             raise KedroCliError(
73                 f"'{pipe}' pipeline not found! Existing pipelines: {existing_pls}"
74             )
75 
76         unused_ds = catalog_ds - pipeline_ds
77         default_ds = pipeline_ds - catalog_ds
78         used_ds = catalog_ds - unused_ds
79 
80         # resolve any factory datasets in the pipeline
81         factory_ds_by_type = defaultdict(list)
82         for ds_name in default_ds:
83             matched_pattern = data_catalog._match_pattern(
84                 data_catalog._dataset_patterns, ds_name
85             )
86             if matched_pattern:
87                 ds_config = data_catalog._resolve_config(ds_name, matched_pattern)
88                 factory_ds_by_type[ds_config["type"]].append(ds_name)
89 
90         default_ds = default_ds - set(chain.from_iterable(factory_ds_by_type.values()))
91 
92         unused_by_type = _map_type_to_datasets(unused_ds, datasets_meta)
93         used_by_type = _map_type_to_datasets(used_ds, datasets_meta)
94 
95         if default_ds:
96             used_by_type["DefaultDataset"].extend(default_ds)
97 
98         data = (
99             (mentioned, dict(used_by_type)),
100             (factories, dict(factory_ds_by_type)),
101             (not_mentioned, dict(unused_by_type)),
102         )
103         result[title.format(pipe)] = {key: value for key, value in data if value}
104     secho(yaml.dump(result))
105 
106 
107 def _map_type_to_datasets(datasets, datasets_meta):
108     """Build dictionary with a dataset type as a key and list of
109     datasets of the specific type as a value.
110     """
111     mapping = defaultdict(list)
112     for dataset in datasets:
113         is_param = dataset.startswith("params:") or dataset == "parameters"
114         if not is_param:
115             ds_type = datasets_meta[dataset].__class__.__name__
116             if dataset not in mapping[ds_type]:
117                 mapping[ds_type].append(dataset)
118     return mapping
119 
120 
121 @catalog.command("create")
122 @env_option(help="Environment to create Data Catalog YAML file in. Defaults to `base`.")
123 @click.option(
124     "--pipeline",
125     "-p",
126     "pipeline_name",
127     type=str,
128     required=True,
129     help="Name of a pipeline.",
130 )
131 @click.pass_obj
132 def create_catalog(metadata: ProjectMetadata, pipeline_name, env):
133     """Create Data Catalog YAML configuration with missing datasets.
134 
135     Add ``MemoryDataset`` datasets to Data Catalog YAML configuration
136     file for each dataset in a registered pipeline if it is missing from
137     the ``DataCatalog``.
138 
139     The catalog configuration will be saved to
140     `<conf_source>/<env>/catalog/<pipeline_name>.yml` file.
141     """
142     env = env or "base"
143     session = _create_session(metadata.package_name, env=env)
144     context = session.load_context()
145 
146     pipeline = pipelines.get(pipeline_name)
147 
148     if not pipeline:
149         existing_pipelines = ", ".join(sorted(pipelines.keys()))
150         raise KedroCliError(
151             f"'{pipeline_name}' pipeline not found! Existing pipelines: {existing_pipelines}"
152         )
153 
154     pipe_datasets = {
155         ds_name
156         for ds_name in pipeline.data_sets()
157         if not ds_name.startswith("params:") and ds_name != "parameters"
158     }
159 
160     catalog_datasets = {
161         ds_name
162         for ds_name in context.catalog._data_sets.keys()  # noqa: protected-access
163         if not ds_name.startswith("params:") and ds_name != "parameters"
164     }
165 
166     # Datasets that are missing in Data Catalog
167     missing_ds = sorted(pipe_datasets - catalog_datasets)
168     if missing_ds:
169         catalog_path = (
170             context.project_path
171             / settings.CONF_SOURCE
172             / env
173             / f"catalog_{pipeline_name}.yml"
174         )
175         _add_missing_datasets_to_catalog(missing_ds, catalog_path)
176         click.echo(f"Data Catalog YAML configuration was created: {catalog_path}")
177     else:
178         click.echo("All datasets are already configured.")
179 
180 
181 def _add_missing_datasets_to_catalog(missing_ds, catalog_path):
182     if catalog_path.is_file():
183         catalog_config = yaml.safe_load(catalog_path.read_text()) or {}
184     else:
185         catalog_config = {}
186 
187     for ds_name in missing_ds:
188         catalog_config[ds_name] = {"type": "MemoryDataset"}
189 
190     # Create only `catalog` folder under existing environment
191     # (all parent folders must exist).
192     catalog_path.parent.mkdir(exist_ok=True)
193     with catalog_path.open(mode="w") as catalog_file:
194         yaml.safe_dump(catalog_config, catalog_file, default_flow_style=False)
195 
196 
197 @catalog.command("rank")
198 @env_option
199 @click.pass_obj
200 def rank_catalog_factories(metadata: ProjectMetadata, env):
201     """List all dataset factories in the catalog, ranked by priority by which they are matched."""
202     session = _create_session(metadata.package_name, env=env)
203     context = session.load_context()
204 
205     catalog_factories = context.catalog._dataset_patterns
206     if catalog_factories:
207         click.echo(yaml.dump(list(catalog_factories.keys())))
208     else:
209         click.echo("There are no dataset factories in the catalog.")
210 
211 
212 @catalog.command("resolve")
213 @env_option
214 @click.pass_obj
215 def resolve_patterns(metadata: ProjectMetadata, env):
216     """Resolve catalog factories against pipeline datasets"""
217 
218     session = _create_session(metadata.package_name, env=env)
219     context = session.load_context()
220 
221     data_catalog = context.catalog
222     catalog_config = context.config_loader["catalog"]
223 
224     explicit_datasets = {
225         ds_name: ds_config
226         for ds_name, ds_config in catalog_config.items()
227         if not data_catalog._is_pattern(ds_name)
228     }
229 
230     target_pipelines = pipelines.keys()
231     datasets = set()
232 
233     for pipe in target_pipelines:
234         pl_obj = pipelines.get(pipe)
235         if pl_obj:
236             datasets.update(pl_obj.data_sets())
237 
238     for ds_name in datasets:
239         is_param = ds_name.startswith("params:") or ds_name == "parameters"
240         if ds_name in explicit_datasets or is_param:
241             continue
242 
243         matched_pattern = data_catalog._match_pattern(
244             data_catalog._dataset_patterns, ds_name
245         )
246         if matched_pattern:
247             ds_config = data_catalog._resolve_config(ds_name, matched_pattern)
248             ds_config["filepath"] = _trim_filepath(
249                 str(context.project_path) + "/", ds_config["filepath"]
250             )
251             explicit_datasets[ds_name] = ds_config
252 
253     secho(yaml.dump(explicit_datasets))
254 
255 
256 def _trim_filepath(project_path: str, file_path: str):
257     return file_path.replace(project_path, "", 1)
258
```
--- 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/kedro/framework/cli/catalog.py b/kedro/framework/cli/catalog.py
--- a/kedro/framework/cli/catalog.py
+++ b/kedro/framework/cli/catalog.py
@@ -137,7 +137,7 @@
     the ``DataCatalog``.
 
     The catalog configuration will be saved to
-    `<conf_source>/<env>/catalog/<pipeline_name>.yml` file.
+    `<conf_source>/<env>/catalog_<pipeline_name>.yml` file.
     """
     env = env or "base"
     session = _create_session(metadata.package_name, env=env)

{"golden_diff": "diff --git a/kedro/framework/cli/catalog.py b/kedro/framework/cli/catalog.py\n--- a/kedro/framework/cli/catalog.py\n+++ b/kedro/framework/cli/catalog.py\n@@ -137,7 +137,7 @@\n     the ``DataCatalog``.\n \n     The catalog configuration will be saved to\n-    `<conf_source>/<env>/catalog/<pipeline_name>.yml` file.\n+    `<conf_source>/<env>/catalog_<pipeline_name>.yml` file.\n     \"\"\"\n     env = env or \"base\"\n     session = _create_session(metadata.package_name, env=env)\n", "issue": "Final clean up for `kedro pipeline create` and `kedro catalog create` docs\n## Description\r\n<!-- Is your feature request related to a problem? A clear and concise description of what the problem is: \"I'm always frustrated when ...\" -->\r\nSteps to reprdouce:\r\n```\r\nkedro new -s spaceflights --checkout main \r\nkedro pipeline create new\r\n```\r\n\r\n## Inconsistency 1 - `kedro pipeline create` add `README.md` while starter doesn't have it\r\n![image](https://github.com/kedro-org/kedro/assets/18221871/f8b75c44-4613-4769-8f06-5a75f0e633c7)\r\n\r\n## Inconsistency 2 - CLI docstring seems to be outdated.\r\n\r\nhttps://github.com/kedro-org/kedro/blob/6888001c6019059ae717c99fbe26a06f67d73ca2/kedro/framework/cli/catalog.py#L131C1-L141C8\r\n\r\n## Inconsistency 3 - broken doc\r\nIt generates a link that doesn't exist - https://docs.kedro.org/en/0.18.12/kedro_project_setup/configuration.html#parameters\r\n\r\nSee this:\r\nhttps://github.com/kedro-org/kedro/blob/6888001c6019059ae717c99fbe26a06f67d73ca2/kedro/templates/pipeline/%7B%7B%20cookiecutter.pipeline_name%20%7D%7D/config/parameters_%7B%7B%20cookiecutter.pipeline_name%20%7D%7D.yml#L5\r\n\r\nAlso, any idea it is in a folder called `config` but not `conf`?\r\n## Outdated Doc 4\r\nSee https://github.com/kedro-org/kedro/pull/2888#discussion_r1297170881\r\n\r\n\r\n\r\n## Context\r\n<!-- Why is this change important to you? How would you use it? How can it benefit other users? -->\r\n\r\n## Possible Implementation\r\n<!-- (Optional) Suggest an idea for implementing the addition or change. -->\r\n\r\n## Possible Alternatives\r\n<!-- (Optional) Describe any alternative solutions or features you've considered. -->\r\n\n", "before_files": [{"content": "\"\"\"A collection of CLI commands for working with Kedro catalog.\"\"\"\nfrom collections import defaultdict\nfrom itertools import chain\n\nimport click\nimport yaml\nfrom click import secho\n\nfrom kedro.framework.cli.utils import KedroCliError, env_option, split_string\nfrom kedro.framework.project import pipelines, settings\nfrom kedro.framework.session import KedroSession\nfrom kedro.framework.startup import ProjectMetadata\n\n\ndef _create_session(package_name: str, **kwargs):\n    kwargs.setdefault(\"save_on_close\", False)\n    try:\n        return KedroSession.create(package_name, **kwargs)\n    except Exception as exc:\n        raise KedroCliError(\n            f\"Unable to instantiate Kedro session.\\nError: {exc}\"\n        ) from exc\n\n\n# noqa: missing-function-docstring\n@click.group(name=\"Kedro\")\ndef catalog_cli():  # pragma: no cover\n    pass\n\n\n@catalog_cli.group()\ndef catalog():\n    \"\"\"Commands for working with catalog.\"\"\"\n\n\n# noqa: too-many-locals,protected-access\n@catalog.command(\"list\")\n@env_option\n@click.option(\n    \"--pipeline\",\n    \"-p\",\n    type=str,\n    default=\"\",\n    help=\"Name of the modular pipeline to run. If not set, \"\n    \"the project pipeline is run by default.\",\n    callback=split_string,\n)\n@click.pass_obj\ndef list_datasets(metadata: ProjectMetadata, pipeline, env):\n    \"\"\"Show datasets per type.\"\"\"\n    title = \"Datasets in '{}' pipeline\"\n    not_mentioned = \"Datasets not mentioned in pipeline\"\n    mentioned = \"Datasets mentioned in pipeline\"\n    factories = \"Datasets generated from factories\"\n\n    session = _create_session(metadata.package_name, env=env)\n    context = session.load_context()\n\n    data_catalog = context.catalog\n    datasets_meta = data_catalog._data_sets\n    catalog_ds = set(data_catalog.list())\n\n    target_pipelines = pipeline or pipelines.keys()\n\n    result = {}\n    for pipe in target_pipelines:\n        pl_obj = pipelines.get(pipe)\n        if pl_obj:\n            pipeline_ds = pl_obj.data_sets()\n        else:\n            existing_pls = \", \".join(sorted(pipelines.keys()))\n            raise KedroCliError(\n                f\"'{pipe}' pipeline not found! Existing pipelines: {existing_pls}\"\n            )\n\n        unused_ds = catalog_ds - pipeline_ds\n        default_ds = pipeline_ds - catalog_ds\n        used_ds = catalog_ds - unused_ds\n\n        # resolve any factory datasets in the pipeline\n        factory_ds_by_type = defaultdict(list)\n        for ds_name in default_ds:\n            matched_pattern = data_catalog._match_pattern(\n                data_catalog._dataset_patterns, ds_name\n            )\n            if matched_pattern:\n                ds_config = data_catalog._resolve_config(ds_name, matched_pattern)\n                factory_ds_by_type[ds_config[\"type\"]].append(ds_name)\n\n        default_ds = default_ds - set(chain.from_iterable(factory_ds_by_type.values()))\n\n        unused_by_type = _map_type_to_datasets(unused_ds, datasets_meta)\n        used_by_type = _map_type_to_datasets(used_ds, datasets_meta)\n\n        if default_ds:\n            used_by_type[\"DefaultDataset\"].extend(default_ds)\n\n        data = (\n            (mentioned, dict(used_by_type)),\n            (factories, dict(factory_ds_by_type)),\n            (not_mentioned, dict(unused_by_type)),\n        )\n        result[title.format(pipe)] = {key: value for key, value in data if value}\n    secho(yaml.dump(result))\n\n\ndef _map_type_to_datasets(datasets, datasets_meta):\n    \"\"\"Build dictionary with a dataset type as a key and list of\n    datasets of the specific type as a value.\n    \"\"\"\n    mapping = defaultdict(list)\n    for dataset in datasets:\n        is_param = dataset.startswith(\"params:\") or dataset == \"parameters\"\n        if not is_param:\n            ds_type = datasets_meta[dataset].__class__.__name__\n            if dataset not in mapping[ds_type]:\n                mapping[ds_type].append(dataset)\n    return mapping\n\n\n@catalog.command(\"create\")\n@env_option(help=\"Environment to create Data Catalog YAML file in. Defaults to `base`.\")\n@click.option(\n    \"--pipeline\",\n    \"-p\",\n    \"pipeline_name\",\n    type=str,\n    required=True,\n    help=\"Name of a pipeline.\",\n)\n@click.pass_obj\ndef create_catalog(metadata: ProjectMetadata, pipeline_name, env):\n    \"\"\"Create Data Catalog YAML configuration with missing datasets.\n\n    Add ``MemoryDataset`` datasets to Data Catalog YAML configuration\n    file for each dataset in a registered pipeline if it is missing from\n    the ``DataCatalog``.\n\n    The catalog configuration will be saved to\n    `<conf_source>/<env>/catalog/<pipeline_name>.yml` file.\n    \"\"\"\n    env = env or \"base\"\n    session = _create_session(metadata.package_name, env=env)\n    context = session.load_context()\n\n    pipeline = pipelines.get(pipeline_name)\n\n    if not pipeline:\n        existing_pipelines = \", \".join(sorted(pipelines.keys()))\n        raise KedroCliError(\n            f\"'{pipeline_name}' pipeline not found! Existing pipelines: {existing_pipelines}\"\n        )\n\n    pipe_datasets = {\n        ds_name\n        for ds_name in pipeline.data_sets()\n        if not ds_name.startswith(\"params:\") and ds_name != \"parameters\"\n    }\n\n    catalog_datasets = {\n        ds_name\n        for ds_name in context.catalog._data_sets.keys()  # noqa: protected-access\n        if not ds_name.startswith(\"params:\") and ds_name != \"parameters\"\n    }\n\n    # Datasets that are missing in Data Catalog\n    missing_ds = sorted(pipe_datasets - catalog_datasets)\n    if missing_ds:\n        catalog_path = (\n            context.project_path\n            / settings.CONF_SOURCE\n            / env\n            / f\"catalog_{pipeline_name}.yml\"\n        )\n        _add_missing_datasets_to_catalog(missing_ds, catalog_path)\n        click.echo(f\"Data Catalog YAML configuration was created: {catalog_path}\")\n    else:\n        click.echo(\"All datasets are already configured.\")\n\n\ndef _add_missing_datasets_to_catalog(missing_ds, catalog_path):\n    if catalog_path.is_file():\n        catalog_config = yaml.safe_load(catalog_path.read_text()) or {}\n    else:\n        catalog_config = {}\n\n    for ds_name in missing_ds:\n        catalog_config[ds_name] = {\"type\": \"MemoryDataset\"}\n\n    # Create only `catalog` folder under existing environment\n    # (all parent folders must exist).\n    catalog_path.parent.mkdir(exist_ok=True)\n    with catalog_path.open(mode=\"w\") as catalog_file:\n        yaml.safe_dump(catalog_config, catalog_file, default_flow_style=False)\n\n\n@catalog.command(\"rank\")\n@env_option\n@click.pass_obj\ndef rank_catalog_factories(metadata: ProjectMetadata, env):\n    \"\"\"List all dataset factories in the catalog, ranked by priority by which they are matched.\"\"\"\n    session = _create_session(metadata.package_name, env=env)\n    context = session.load_context()\n\n    catalog_factories = context.catalog._dataset_patterns\n    if catalog_factories:\n        click.echo(yaml.dump(list(catalog_factories.keys())))\n    else:\n        click.echo(\"There are no dataset factories in the catalog.\")\n\n\n@catalog.command(\"resolve\")\n@env_option\n@click.pass_obj\ndef resolve_patterns(metadata: ProjectMetadata, env):\n    \"\"\"Resolve catalog factories against pipeline datasets\"\"\"\n\n    session = _create_session(metadata.package_name, env=env)\n    context = session.load_context()\n\n    data_catalog = context.catalog\n    catalog_config = context.config_loader[\"catalog\"]\n\n    explicit_datasets = {\n        ds_name: ds_config\n        for ds_name, ds_config in catalog_config.items()\n        if not data_catalog._is_pattern(ds_name)\n    }\n\n    target_pipelines = pipelines.keys()\n    datasets = set()\n\n    for pipe in target_pipelines:\n        pl_obj = pipelines.get(pipe)\n        if pl_obj:\n            datasets.update(pl_obj.data_sets())\n\n    for ds_name in datasets:\n        is_param = ds_name.startswith(\"params:\") or ds_name == \"parameters\"\n        if ds_name in explicit_datasets or is_param:\n            continue\n\n        matched_pattern = data_catalog._match_pattern(\n            data_catalog._dataset_patterns, ds_name\n        )\n        if matched_pattern:\n            ds_config = data_catalog._resolve_config(ds_name, matched_pattern)\n            ds_config[\"filepath\"] = _trim_filepath(\n                str(context.project_path) + \"/\", ds_config[\"filepath\"]\n            )\n            explicit_datasets[ds_name] = ds_config\n\n    secho(yaml.dump(explicit_datasets))\n\n\ndef _trim_filepath(project_path: str, file_path: str):\n    return file_path.replace(project_path, \"\", 1)\n", "path": "kedro/framework/cli/catalog.py"}], "after_files": [{"content": "\"\"\"A collection of CLI commands for working with Kedro catalog.\"\"\"\nfrom collections import defaultdict\nfrom itertools import chain\n\nimport click\nimport yaml\nfrom click import secho\n\nfrom kedro.framework.cli.utils import KedroCliError, env_option, split_string\nfrom kedro.framework.project import pipelines, settings\nfrom kedro.framework.session import KedroSession\nfrom kedro.framework.startup import ProjectMetadata\n\n\ndef _create_session(package_name: str, **kwargs):\n    kwargs.setdefault(\"save_on_close\", False)\n    try:\n        return KedroSession.create(package_name, **kwargs)\n    except Exception as exc:\n        raise KedroCliError(\n            f\"Unable to instantiate Kedro session.\\nError: {exc}\"\n        ) from exc\n\n\n# noqa: missing-function-docstring\n@click.group(name=\"Kedro\")\ndef catalog_cli():  # pragma: no cover\n    pass\n\n\n@catalog_cli.group()\ndef catalog():\n    \"\"\"Commands for working with catalog.\"\"\"\n\n\n# noqa: too-many-locals,protected-access\n@catalog.command(\"list\")\n@env_option\n@click.option(\n    \"--pipeline\",\n    \"-p\",\n    type=str,\n    default=\"\",\n    help=\"Name of the modular pipeline to run. If not set, \"\n    \"the project pipeline is run by default.\",\n    callback=split_string,\n)\n@click.pass_obj\ndef list_datasets(metadata: ProjectMetadata, pipeline, env):\n    \"\"\"Show datasets per type.\"\"\"\n    title = \"Datasets in '{}' pipeline\"\n    not_mentioned = \"Datasets not mentioned in pipeline\"\n    mentioned = \"Datasets mentioned in pipeline\"\n    factories = \"Datasets generated from factories\"\n\n    session = _create_session(metadata.package_name, env=env)\n    context = session.load_context()\n\n    data_catalog = context.catalog\n    datasets_meta = data_catalog._data_sets\n    catalog_ds = set(data_catalog.list())\n\n    target_pipelines = pipeline or pipelines.keys()\n\n    result = {}\n    for pipe in target_pipelines:\n        pl_obj = pipelines.get(pipe)\n        if pl_obj:\n            pipeline_ds = pl_obj.data_sets()\n        else:\n            existing_pls = \", \".join(sorted(pipelines.keys()))\n            raise KedroCliError(\n                f\"'{pipe}' pipeline not found! Existing pipelines: {existing_pls}\"\n            )\n\n        unused_ds = catalog_ds - pipeline_ds\n        default_ds = pipeline_ds - catalog_ds\n        used_ds = catalog_ds - unused_ds\n\n        # resolve any factory datasets in the pipeline\n        factory_ds_by_type = defaultdict(list)\n        for ds_name in default_ds:\n            matched_pattern = data_catalog._match_pattern(\n                data_catalog._dataset_patterns, ds_name\n            )\n            if matched_pattern:\n                ds_config = data_catalog._resolve_config(ds_name, matched_pattern)\n                factory_ds_by_type[ds_config[\"type\"]].append(ds_name)\n\n        default_ds = default_ds - set(chain.from_iterable(factory_ds_by_type.values()))\n\n        unused_by_type = _map_type_to_datasets(unused_ds, datasets_meta)\n        used_by_type = _map_type_to_datasets(used_ds, datasets_meta)\n\n        if default_ds:\n            used_by_type[\"DefaultDataset\"].extend(default_ds)\n\n        data = (\n            (mentioned, dict(used_by_type)),\n            (factories, dict(factory_ds_by_type)),\n            (not_mentioned, dict(unused_by_type)),\n        )\n        result[title.format(pipe)] = {key: value for key, value in data if value}\n    secho(yaml.dump(result))\n\n\ndef _map_type_to_datasets(datasets, datasets_meta):\n    \"\"\"Build dictionary with a dataset type as a key and list of\n    datasets of the specific type as a value.\n    \"\"\"\n    mapping = defaultdict(list)\n    for dataset in datasets:\n        is_param = dataset.startswith(\"params:\") or dataset == \"parameters\"\n        if not is_param:\n            ds_type = datasets_meta[dataset].__class__.__name__\n            if dataset not in mapping[ds_type]:\n                mapping[ds_type].append(dataset)\n    return mapping\n\n\n@catalog.command(\"create\")\n@env_option(help=\"Environment to create Data Catalog YAML file in. Defaults to `base`.\")\n@click.option(\n    \"--pipeline\",\n    \"-p\",\n    \"pipeline_name\",\n    type=str,\n    required=True,\n    help=\"Name of a pipeline.\",\n)\n@click.pass_obj\ndef create_catalog(metadata: ProjectMetadata, pipeline_name, env):\n    \"\"\"Create Data Catalog YAML configuration with missing datasets.\n\n    Add ``MemoryDataset`` datasets to Data Catalog YAML configuration\n    file for each dataset in a registered pipeline if it is missing from\n    the ``DataCatalog``.\n\n    The catalog configuration will be saved to\n    `<conf_source>/<env>/catalog_<pipeline_name>.yml` file.\n    \"\"\"\n    env = env or \"base\"\n    session = _create_session(metadata.package_name, env=env)\n    context = session.load_context()\n\n    pipeline = pipelines.get(pipeline_name)\n\n    if not pipeline:\n        existing_pipelines = \", \".join(sorted(pipelines.keys()))\n        raise KedroCliError(\n            f\"'{pipeline_name}' pipeline not found! Existing pipelines: {existing_pipelines}\"\n        )\n\n    pipe_datasets = {\n        ds_name\n        for ds_name in pipeline.data_sets()\n        if not ds_name.startswith(\"params:\") and ds_name != \"parameters\"\n    }\n\n    catalog_datasets = {\n        ds_name\n        for ds_name in context.catalog._data_sets.keys()  # noqa: protected-access\n        if not ds_name.startswith(\"params:\") and ds_name != \"parameters\"\n    }\n\n    # Datasets that are missing in Data Catalog\n    missing_ds = sorted(pipe_datasets - catalog_datasets)\n    if missing_ds:\n        catalog_path = (\n            context.project_path\n            / settings.CONF_SOURCE\n            / env\n            / f\"catalog_{pipeline_name}.yml\"\n        )\n        _add_missing_datasets_to_catalog(missing_ds, catalog_path)\n        click.echo(f\"Data Catalog YAML configuration was created: {catalog_path}\")\n    else:\n        click.echo(\"All datasets are already configured.\")\n\n\ndef _add_missing_datasets_to_catalog(missing_ds, catalog_path):\n    if catalog_path.is_file():\n        catalog_config = yaml.safe_load(catalog_path.read_text()) or {}\n    else:\n        catalog_config = {}\n\n    for ds_name in missing_ds:\n        catalog_config[ds_name] = {\"type\": \"MemoryDataset\"}\n\n    # Create only `catalog` folder under existing environment\n    # (all parent folders must exist).\n    catalog_path.parent.mkdir(exist_ok=True)\n    with catalog_path.open(mode=\"w\") as catalog_file:\n        yaml.safe_dump(catalog_config, catalog_file, default_flow_style=False)\n\n\n@catalog.command(\"rank\")\n@env_option\n@click.pass_obj\ndef rank_catalog_factories(metadata: ProjectMetadata, env):\n    \"\"\"List all dataset factories in the catalog, ranked by priority by which they are matched.\"\"\"\n    session = _create_session(metadata.package_name, env=env)\n    context = session.load_context()\n\n    catalog_factories = context.catalog._dataset_patterns\n    if catalog_factories:\n        click.echo(yaml.dump(list(catalog_factories.keys())))\n    else:\n        click.echo(\"There are no dataset factories in the catalog.\")\n\n\n@catalog.command(\"resolve\")\n@env_option\n@click.pass_obj\ndef resolve_patterns(metadata: ProjectMetadata, env):\n    \"\"\"Resolve catalog factories against pipeline datasets\"\"\"\n\n    session = _create_session(metadata.package_name, env=env)\n    context = session.load_context()\n\n    data_catalog = context.catalog\n    catalog_config = context.config_loader[\"catalog\"]\n\n    explicit_datasets = {\n        ds_name: ds_config\n        for ds_name, ds_config in catalog_config.items()\n        if not data_catalog._is_pattern(ds_name)\n    }\n\n    target_pipelines = pipelines.keys()\n    datasets = set()\n\n    for pipe in target_pipelines:\n        pl_obj = pipelines.get(pipe)\n        if pl_obj:\n            datasets.update(pl_obj.data_sets())\n\n    for ds_name in datasets:\n        is_param = ds_name.startswith(\"params:\") or ds_name == \"parameters\"\n        if ds_name in explicit_datasets or is_param:\n            continue\n\n        matched_pattern = data_catalog._match_pattern(\n            data_catalog._dataset_patterns, ds_name\n        )\n        if matched_pattern:\n            ds_config = data_catalog._resolve_config(ds_name, matched_pattern)\n            ds_config[\"filepath\"] = _trim_filepath(\n                str(context.project_path) + \"/\", ds_config[\"filepath\"]\n            )\n            explicit_datasets[ds_name] = ds_config\n\n    secho(yaml.dump(explicit_datasets))\n\n\ndef _trim_filepath(project_path: str, file_path: str):\n    return file_path.replace(project_path, \"\", 1)\n", "path": "kedro/framework/cli/catalog.py"}]}