We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
bug: Cloud CLI schedule help has same text for describe and list
### Meltano Version

2.18.0

### Python Version

3.10

### Bug scope

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

### Operating System

Mac M1

### Description

```
❯ meltano-cloud schedule --help
Usage: meltano-cloud schedule [OPTIONS] COMMAND [ARGS]...

  Interact with Meltano Cloud project schedules.

Options:
  --deployment TEXT  The name of the Meltano Cloud deployment the schedule
                     belongs to.
  --schedule TEXT    The name of a schedule within the specified deployment.
  --help             Show this message and exit.

Commands:
  describe  List Meltano Cloud schedules.
  disable   Disable a Meltano Cloud schedule.
  enable    Enable a Meltano Cloud schedule.
  list      List Meltano Cloud schedules.
```

`describe` and `list` say the same thing

### 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/cloud-cli/meltano/cloud/cli/schedule.py`
Content:
```
1 """Meltano Cloud `schedules` command."""
2 
3 from __future__ import annotations
4 
5 import itertools as it
6 import json
7 import typing as t
8 from datetime import datetime, timedelta, timezone
9 from http import HTTPStatus
10 
11 import click
12 import tabulate
13 from croniter import croniter, croniter_range
14 
15 from meltano.cloud.api.client import MeltanoCloudClient, MeltanoCloudError
16 from meltano.cloud.cli.base import pass_context, run_async, shared_option
17 
18 if t.TYPE_CHECKING:
19     from meltano.cloud.api.config import MeltanoCloudConfig
20     from meltano.cloud.api.types import CloudProjectSchedule
21     from meltano.cloud.cli.base import MeltanoCloudCLIContext
22 
23 MAX_PAGE_SIZE = 250
24 
25 
26 class SchedulesCloudClient(MeltanoCloudClient):
27     """A Meltano Cloud client with extensions for schedules."""
28 
29     async def set_schedule_enabled(
30         self,
31         *,
32         deployment_name: str,
33         schedule_name: str,
34         enabled: bool,
35     ):
36         """Use PUT to update the enabled state of a Meltano Cloud project schedule.
37 
38         Args:
39             deployment: The name of the deployment the schedule belongs to.
40             schedule: The name of the schedule to enable/disable.
41             enabled: Whether the schedule should be enabled.
42 
43         Raises:
44             MeltanoCloudError: The Meltano Cloud API responded with an error.
45         """
46         async with self.authenticated():
47             try:
48                 await self._json_request(
49                     "PUT",
50                     "/".join(
51                         (
52                             "/schedules/v1",
53                             f"{self.config.tenant_resource_key}",
54                             f"{self.config.internal_project_id}",
55                             deployment_name,
56                             schedule_name,
57                             "enabled",
58                         ),
59                     ),
60                     json=enabled,
61                 )
62             except MeltanoCloudError as ex:
63                 if ex.response.status == HTTPStatus.NOT_FOUND:
64                     ex.response.reason = (
65                         f"Unable to find schedule named {schedule_name!r} "
66                         f"within a deployment named {deployment_name!r}"
67                     )
68                     raise MeltanoCloudError(ex.response) from ex
69                 raise
70 
71     async def get_schedule(
72         self,
73         *,
74         deployment_name: str,
75         schedule_name: str,
76     ):
77         """Use GET to get a Meltano Cloud project schedule.
78 
79         Args:
80             deployment_name: The name of the deployment the schedule belongs to.
81             schedule_name: The name of the schedule.
82 
83         Raises:
84             MeltanoCloudError: The Meltano Cloud API responded with an error.
85         """
86         async with self.authenticated():
87             try:
88                 return await self._json_request(
89                     "GET",
90                     "/".join(
91                         (
92                             "/schedules/v1",
93                             self.config.tenant_resource_key,
94                             self.config.internal_project_id,
95                             deployment_name,
96                             schedule_name,
97                         ),
98                     ),
99                 )
100             except MeltanoCloudError as ex:
101                 if ex.response.status == HTTPStatus.NOT_FOUND:
102                     ex.response.reason = (
103                         f"Unable to find schedule named {schedule_name!r} "
104                         f"within a deployment named {deployment_name!r}"
105                     )
106                     raise MeltanoCloudError(ex.response) from ex
107                 raise
108 
109     async def get_schedules(
110         self,
111         *,
112         deployment_name: str | None = None,
113         page_size: int | None = None,
114         page_token: str | None = None,
115     ):
116         """Use GET to get Meltano Cloud project schedules.
117 
118         Args:
119             deployment_name: The name of the deployment the schedule belongs to.
120             page_size: The number of items to request per page.
121             page_token: The page token.
122         """
123         params: dict[str, t.Any] = {
124             "page_size": page_size,
125             "page_token": page_token,
126         }
127 
128         path = [
129             "/schedules/v1",
130             self.config.tenant_resource_key,
131             self.config.internal_project_id,
132         ]
133         if deployment_name is not None:
134             path.append(deployment_name)
135 
136         async with self.authenticated():
137             return await self._json_request(
138                 "GET",
139                 "/".join(path),
140                 params=self.clean_params(params),
141             )
142 
143 
144 async def _set_enabled_state(
145     *,
146     config: MeltanoCloudConfig,
147     deployment_name: str | None,
148     schedule_name: str | None,
149     enabled: bool,
150 ):
151     if schedule_name is None:
152         raise click.UsageError("Missing option '--schedule'")
153     if deployment_name is None:
154         raise click.UsageError("Missing option '--deployment'")
155     async with SchedulesCloudClient(config=config) as client:
156         await client.set_schedule_enabled(
157             deployment_name=deployment_name,
158             schedule_name=schedule_name,
159             enabled=enabled,
160         )
161 
162 
163 deployment_option = shared_option(
164     "--deployment",
165     help="The name of the Meltano Cloud deployment the schedule belongs to.",
166 )
167 
168 schedule_option = shared_option(
169     "--schedule",
170     help="The name of a schedule within the specified deployment.",
171 )
172 
173 
174 @click.group("schedule")
175 @deployment_option
176 @schedule_option
177 def schedule_group() -> None:
178     """Interact with Meltano Cloud project schedules."""
179 
180 
181 @schedule_group.command("enable")
182 @deployment_option
183 @schedule_option
184 @pass_context
185 @run_async
186 async def enable(context: MeltanoCloudCLIContext) -> None:
187     """Enable a Meltano Cloud schedule."""
188     await _set_enabled_state(
189         config=context.config,
190         deployment_name=context.deployment,
191         schedule_name=context.schedule,
192         enabled=True,
193     )
194 
195 
196 @schedule_group.command("disable")
197 @deployment_option
198 @schedule_option
199 @pass_context
200 @run_async
201 async def disable(context: MeltanoCloudCLIContext) -> None:
202     """Disable a Meltano Cloud schedule."""
203     await _set_enabled_state(
204         config=context.config,
205         deployment_name=context.deployment,
206         schedule_name=context.schedule,
207         enabled=False,
208     )
209 
210 
211 async def _get_schedule(
212     config: MeltanoCloudConfig,
213     deployment_name: str | None,
214     schedule_name: str | None,
215 ) -> CloudProjectSchedule:
216     if schedule_name is None:
217         raise click.UsageError("Missing option '--schedule'")
218     if deployment_name is None:
219         raise click.UsageError("Missing option '--deployment'")
220     async with SchedulesCloudClient(config=config) as client:
221         return await client.get_schedule(
222             deployment_name=deployment_name,
223             schedule_name=schedule_name,
224         )
225 
226 
227 async def _get_schedules(
228     config: MeltanoCloudConfig,
229     deployment_name: str | None,
230     limit: int,
231 ) -> list[CloudProjectSchedule]:
232     page_token = None
233     page_size = min(limit, MAX_PAGE_SIZE)
234     results: list[CloudProjectSchedule] = []
235 
236     async with SchedulesCloudClient(config=config) as client:
237         while True:
238             response = await client.get_schedules(
239                 deployment_name=deployment_name,
240                 page_size=page_size,
241                 page_token=page_token,
242             )
243 
244             results.extend(response["results"])
245 
246             if response["pagination"] and len(results) < limit:
247                 page_token = response["pagination"]["next_page_token"]
248             else:
249                 break
250 
251     return results[:limit]
252 
253 
254 def _next_n_runs(n: int, cron_expr: str) -> tuple[datetime, ...]:
255     now = datetime.now(timezone.utc)
256     return tuple(it.islice(croniter(cron_expr, now, ret_type=datetime), n))
257 
258 
259 timedelta_year = timedelta(days=365)  # noqa: WPS432
260 
261 
262 def _approx_daily_freq(
263     cron_expr: str,
264     sample_period: timedelta = timedelta_year,
265     num_digits_precision: int = 1,
266 ) -> str:
267     now = datetime.now(timezone.utc)
268     num_runs = sum(1 for _ in croniter_range(now, now + sample_period, cron_expr))
269     freq = round(num_runs / sample_period.days, num_digits_precision)
270     return "< 1" if freq < 1 else str(freq)
271 
272 
273 def _process_table_row(
274     schedule: CloudProjectSchedule,
275 ) -> tuple[str, str, str, str, bool]:
276     return (
277         schedule["deployment_name"],
278         schedule["schedule_name"],
279         schedule["interval"],
280         _approx_daily_freq(schedule["interval"]),
281         schedule["enabled"],
282     )
283 
284 
285 def _format_schedules_table(
286     schedules: list[CloudProjectSchedule],
287     table_format: str,
288 ) -> str:
289     """Format the schedules as a table.
290 
291     Args:
292         schedules: The schedules to format.
293 
294     Returns:
295         The formatted schedules.
296     """
297     return tabulate.tabulate(
298         [_process_table_row(schedule) for schedule in schedules],
299         headers=(
300             "Deployment",
301             "Schedule",
302             "Interval",
303             "Runs / Day",
304             "Enabled",
305         ),
306         tablefmt=table_format,
307         floatfmt=".1f",
308         # To avoid a tabulate bug (IndexError), only set colalign if there are schedules
309         colalign=("left", "left", "left", "right", "left") if schedules else (),
310     )
311 
312 
313 schedule_list_formatters = {
314     "json": lambda schedules: json.dumps(schedules, indent=2),
315     "markdown": lambda schedules: _format_schedules_table(
316         schedules,
317         table_format="github",
318     ),
319     "terminal": lambda schedules: _format_schedules_table(
320         schedules,
321         table_format="rounded_outline",
322     ),
323 }
324 
325 
326 @schedule_group.command("list")
327 @deployment_option
328 @click.option(
329     "--limit",
330     required=False,
331     type=int,
332     default=10,
333     help="The maximum number of schedules to display.",
334 )
335 @click.option(
336     "--format",
337     "output_format",
338     required=False,
339     default="terminal",
340     type=click.Choice(("terminal", "markdown", "json")),
341     help="The output format to use.",
342 )
343 @pass_context
344 @run_async
345 async def list_schedules(
346     context: MeltanoCloudCLIContext,
347     output_format: str,
348     limit: int,
349 ) -> None:
350     """List Meltano Cloud schedules."""
351     click.echo(
352         schedule_list_formatters[output_format](
353             await _get_schedules(
354                 config=context.config,
355                 deployment_name=context.deployment,
356                 limit=limit,
357             ),
358         ),
359     )
360 
361 
362 @schedule_group.command("describe")
363 @deployment_option
364 @schedule_option
365 @click.option(
366     "--only-upcoming",
367     is_flag=True,
368     help="Only list upcoming scheduled run start datetimes",
369 )
370 @click.option(
371     "--num-upcoming",
372     type=int,
373     default=10,
374     help="The number of upcoming scheduled run start datetimes to list",
375 )
376 @pass_context
377 @run_async
378 async def describe_schedule(
379     context: MeltanoCloudCLIContext,
380     only_upcoming: bool,
381     num_upcoming: int,
382 ) -> None:
383     """List Meltano Cloud schedules."""
384     schedule = await _get_schedule(
385         config=context.config,
386         deployment_name=context.deployment,
387         schedule_name=context.schedule,
388     )
389 
390     if not only_upcoming:
391         click.echo(
392             f"Deployment name: {schedule['deployment_name']}\n"
393             f"Schedule name:   {schedule['schedule_name']}\n"
394             f"Interval:        {schedule['interval']}\n"
395             f"Enabled:         {schedule['enabled']}",
396         )
397         if schedule["enabled"]:
398             click.echo(
399                 "\nApproximate starting date and time (UTC) of "
400                 f"next {num_upcoming} scheduled runs:",
401             )
402     if schedule["enabled"]:
403         for dt in _next_n_runs(num_upcoming, schedule["interval"]):
404             click.echo(dt.strftime("%Y-%m-%d %H:%M"))
405
```
--- 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/cloud-cli/meltano/cloud/cli/schedule.py b/src/cloud-cli/meltano/cloud/cli/schedule.py
--- a/src/cloud-cli/meltano/cloud/cli/schedule.py
+++ b/src/cloud-cli/meltano/cloud/cli/schedule.py
@@ -380,7 +380,7 @@
     only_upcoming: bool,
     num_upcoming: int,
 ) -> None:
-    """List Meltano Cloud schedules."""
+    """Describe a Meltano Cloud schedules & list upcoming scheduled runs."""
     schedule = await _get_schedule(
         config=context.config,
         deployment_name=context.deployment,

{"golden_diff": "diff --git a/src/cloud-cli/meltano/cloud/cli/schedule.py b/src/cloud-cli/meltano/cloud/cli/schedule.py\n--- a/src/cloud-cli/meltano/cloud/cli/schedule.py\n+++ b/src/cloud-cli/meltano/cloud/cli/schedule.py\n@@ -380,7 +380,7 @@\n     only_upcoming: bool,\n     num_upcoming: int,\n ) -> None:\n-    \"\"\"List Meltano Cloud schedules.\"\"\"\n+    \"\"\"Describe a Meltano Cloud schedules & list upcoming scheduled runs.\"\"\"\n     schedule = await _get_schedule(\n         config=context.config,\n         deployment_name=context.deployment,\n", "issue": "bug: Cloud CLI schedule help has same text for describe and list\n### Meltano Version\n\n2.18.0\n\n### Python Version\n\n3.10\n\n### Bug scope\n\nCLI (options, error messages, logging, etc.)\n\n### Operating System\n\nMac M1\n\n### Description\n\n```\r\n\u276f meltano-cloud schedule --help\r\nUsage: meltano-cloud schedule [OPTIONS] COMMAND [ARGS]...\r\n\r\n  Interact with Meltano Cloud project schedules.\r\n\r\nOptions:\r\n  --deployment TEXT  The name of the Meltano Cloud deployment the schedule\r\n                     belongs to.\r\n  --schedule TEXT    The name of a schedule within the specified deployment.\r\n  --help             Show this message and exit.\r\n\r\nCommands:\r\n  describe  List Meltano Cloud schedules.\r\n  disable   Disable a Meltano Cloud schedule.\r\n  enable    Enable a Meltano Cloud schedule.\r\n  list      List Meltano Cloud schedules.\r\n```\r\n\r\n`describe` and `list` say the same thing\n\n### Code\n\n_No response_\n", "before_files": [{"content": "\"\"\"Meltano Cloud `schedules` command.\"\"\"\n\nfrom __future__ import annotations\n\nimport itertools as it\nimport json\nimport typing as t\nfrom datetime import datetime, timedelta, timezone\nfrom http import HTTPStatus\n\nimport click\nimport tabulate\nfrom croniter import croniter, croniter_range\n\nfrom meltano.cloud.api.client import MeltanoCloudClient, MeltanoCloudError\nfrom meltano.cloud.cli.base import pass_context, run_async, shared_option\n\nif t.TYPE_CHECKING:\n    from meltano.cloud.api.config import MeltanoCloudConfig\n    from meltano.cloud.api.types import CloudProjectSchedule\n    from meltano.cloud.cli.base import MeltanoCloudCLIContext\n\nMAX_PAGE_SIZE = 250\n\n\nclass SchedulesCloudClient(MeltanoCloudClient):\n    \"\"\"A Meltano Cloud client with extensions for schedules.\"\"\"\n\n    async def set_schedule_enabled(\n        self,\n        *,\n        deployment_name: str,\n        schedule_name: str,\n        enabled: bool,\n    ):\n        \"\"\"Use PUT to update the enabled state of a Meltano Cloud project schedule.\n\n        Args:\n            deployment: The name of the deployment the schedule belongs to.\n            schedule: The name of the schedule to enable/disable.\n            enabled: Whether the schedule should be enabled.\n\n        Raises:\n            MeltanoCloudError: The Meltano Cloud API responded with an error.\n        \"\"\"\n        async with self.authenticated():\n            try:\n                await self._json_request(\n                    \"PUT\",\n                    \"/\".join(\n                        (\n                            \"/schedules/v1\",\n                            f\"{self.config.tenant_resource_key}\",\n                            f\"{self.config.internal_project_id}\",\n                            deployment_name,\n                            schedule_name,\n                            \"enabled\",\n                        ),\n                    ),\n                    json=enabled,\n                )\n            except MeltanoCloudError as ex:\n                if ex.response.status == HTTPStatus.NOT_FOUND:\n                    ex.response.reason = (\n                        f\"Unable to find schedule named {schedule_name!r} \"\n                        f\"within a deployment named {deployment_name!r}\"\n                    )\n                    raise MeltanoCloudError(ex.response) from ex\n                raise\n\n    async def get_schedule(\n        self,\n        *,\n        deployment_name: str,\n        schedule_name: str,\n    ):\n        \"\"\"Use GET to get a Meltano Cloud project schedule.\n\n        Args:\n            deployment_name: The name of the deployment the schedule belongs to.\n            schedule_name: The name of the schedule.\n\n        Raises:\n            MeltanoCloudError: The Meltano Cloud API responded with an error.\n        \"\"\"\n        async with self.authenticated():\n            try:\n                return await self._json_request(\n                    \"GET\",\n                    \"/\".join(\n                        (\n                            \"/schedules/v1\",\n                            self.config.tenant_resource_key,\n                            self.config.internal_project_id,\n                            deployment_name,\n                            schedule_name,\n                        ),\n                    ),\n                )\n            except MeltanoCloudError as ex:\n                if ex.response.status == HTTPStatus.NOT_FOUND:\n                    ex.response.reason = (\n                        f\"Unable to find schedule named {schedule_name!r} \"\n                        f\"within a deployment named {deployment_name!r}\"\n                    )\n                    raise MeltanoCloudError(ex.response) from ex\n                raise\n\n    async def get_schedules(\n        self,\n        *,\n        deployment_name: str | None = None,\n        page_size: int | None = None,\n        page_token: str | None = None,\n    ):\n        \"\"\"Use GET to get Meltano Cloud project schedules.\n\n        Args:\n            deployment_name: The name of the deployment the schedule belongs to.\n            page_size: The number of items to request per page.\n            page_token: The page token.\n        \"\"\"\n        params: dict[str, t.Any] = {\n            \"page_size\": page_size,\n            \"page_token\": page_token,\n        }\n\n        path = [\n            \"/schedules/v1\",\n            self.config.tenant_resource_key,\n            self.config.internal_project_id,\n        ]\n        if deployment_name is not None:\n            path.append(deployment_name)\n\n        async with self.authenticated():\n            return await self._json_request(\n                \"GET\",\n                \"/\".join(path),\n                params=self.clean_params(params),\n            )\n\n\nasync def _set_enabled_state(\n    *,\n    config: MeltanoCloudConfig,\n    deployment_name: str | None,\n    schedule_name: str | None,\n    enabled: bool,\n):\n    if schedule_name is None:\n        raise click.UsageError(\"Missing option '--schedule'\")\n    if deployment_name is None:\n        raise click.UsageError(\"Missing option '--deployment'\")\n    async with SchedulesCloudClient(config=config) as client:\n        await client.set_schedule_enabled(\n            deployment_name=deployment_name,\n            schedule_name=schedule_name,\n            enabled=enabled,\n        )\n\n\ndeployment_option = shared_option(\n    \"--deployment\",\n    help=\"The name of the Meltano Cloud deployment the schedule belongs to.\",\n)\n\nschedule_option = shared_option(\n    \"--schedule\",\n    help=\"The name of a schedule within the specified deployment.\",\n)\n\n\n@click.group(\"schedule\")\n@deployment_option\n@schedule_option\ndef schedule_group() -> None:\n    \"\"\"Interact with Meltano Cloud project schedules.\"\"\"\n\n\n@schedule_group.command(\"enable\")\n@deployment_option\n@schedule_option\n@pass_context\n@run_async\nasync def enable(context: MeltanoCloudCLIContext) -> None:\n    \"\"\"Enable a Meltano Cloud schedule.\"\"\"\n    await _set_enabled_state(\n        config=context.config,\n        deployment_name=context.deployment,\n        schedule_name=context.schedule,\n        enabled=True,\n    )\n\n\n@schedule_group.command(\"disable\")\n@deployment_option\n@schedule_option\n@pass_context\n@run_async\nasync def disable(context: MeltanoCloudCLIContext) -> None:\n    \"\"\"Disable a Meltano Cloud schedule.\"\"\"\n    await _set_enabled_state(\n        config=context.config,\n        deployment_name=context.deployment,\n        schedule_name=context.schedule,\n        enabled=False,\n    )\n\n\nasync def _get_schedule(\n    config: MeltanoCloudConfig,\n    deployment_name: str | None,\n    schedule_name: str | None,\n) -> CloudProjectSchedule:\n    if schedule_name is None:\n        raise click.UsageError(\"Missing option '--schedule'\")\n    if deployment_name is None:\n        raise click.UsageError(\"Missing option '--deployment'\")\n    async with SchedulesCloudClient(config=config) as client:\n        return await client.get_schedule(\n            deployment_name=deployment_name,\n            schedule_name=schedule_name,\n        )\n\n\nasync def _get_schedules(\n    config: MeltanoCloudConfig,\n    deployment_name: str | None,\n    limit: int,\n) -> list[CloudProjectSchedule]:\n    page_token = None\n    page_size = min(limit, MAX_PAGE_SIZE)\n    results: list[CloudProjectSchedule] = []\n\n    async with SchedulesCloudClient(config=config) as client:\n        while True:\n            response = await client.get_schedules(\n                deployment_name=deployment_name,\n                page_size=page_size,\n                page_token=page_token,\n            )\n\n            results.extend(response[\"results\"])\n\n            if response[\"pagination\"] and len(results) < limit:\n                page_token = response[\"pagination\"][\"next_page_token\"]\n            else:\n                break\n\n    return results[:limit]\n\n\ndef _next_n_runs(n: int, cron_expr: str) -> tuple[datetime, ...]:\n    now = datetime.now(timezone.utc)\n    return tuple(it.islice(croniter(cron_expr, now, ret_type=datetime), n))\n\n\ntimedelta_year = timedelta(days=365)  # noqa: WPS432\n\n\ndef _approx_daily_freq(\n    cron_expr: str,\n    sample_period: timedelta = timedelta_year,\n    num_digits_precision: int = 1,\n) -> str:\n    now = datetime.now(timezone.utc)\n    num_runs = sum(1 for _ in croniter_range(now, now + sample_period, cron_expr))\n    freq = round(num_runs / sample_period.days, num_digits_precision)\n    return \"< 1\" if freq < 1 else str(freq)\n\n\ndef _process_table_row(\n    schedule: CloudProjectSchedule,\n) -> tuple[str, str, str, str, bool]:\n    return (\n        schedule[\"deployment_name\"],\n        schedule[\"schedule_name\"],\n        schedule[\"interval\"],\n        _approx_daily_freq(schedule[\"interval\"]),\n        schedule[\"enabled\"],\n    )\n\n\ndef _format_schedules_table(\n    schedules: list[CloudProjectSchedule],\n    table_format: str,\n) -> str:\n    \"\"\"Format the schedules as a table.\n\n    Args:\n        schedules: The schedules to format.\n\n    Returns:\n        The formatted schedules.\n    \"\"\"\n    return tabulate.tabulate(\n        [_process_table_row(schedule) for schedule in schedules],\n        headers=(\n            \"Deployment\",\n            \"Schedule\",\n            \"Interval\",\n            \"Runs / Day\",\n            \"Enabled\",\n        ),\n        tablefmt=table_format,\n        floatfmt=\".1f\",\n        # To avoid a tabulate bug (IndexError), only set colalign if there are schedules\n        colalign=(\"left\", \"left\", \"left\", \"right\", \"left\") if schedules else (),\n    )\n\n\nschedule_list_formatters = {\n    \"json\": lambda schedules: json.dumps(schedules, indent=2),\n    \"markdown\": lambda schedules: _format_schedules_table(\n        schedules,\n        table_format=\"github\",\n    ),\n    \"terminal\": lambda schedules: _format_schedules_table(\n        schedules,\n        table_format=\"rounded_outline\",\n    ),\n}\n\n\n@schedule_group.command(\"list\")\n@deployment_option\n@click.option(\n    \"--limit\",\n    required=False,\n    type=int,\n    default=10,\n    help=\"The maximum number of schedules to display.\",\n)\n@click.option(\n    \"--format\",\n    \"output_format\",\n    required=False,\n    default=\"terminal\",\n    type=click.Choice((\"terminal\", \"markdown\", \"json\")),\n    help=\"The output format to use.\",\n)\n@pass_context\n@run_async\nasync def list_schedules(\n    context: MeltanoCloudCLIContext,\n    output_format: str,\n    limit: int,\n) -> None:\n    \"\"\"List Meltano Cloud schedules.\"\"\"\n    click.echo(\n        schedule_list_formatters[output_format](\n            await _get_schedules(\n                config=context.config,\n                deployment_name=context.deployment,\n                limit=limit,\n            ),\n        ),\n    )\n\n\n@schedule_group.command(\"describe\")\n@deployment_option\n@schedule_option\n@click.option(\n    \"--only-upcoming\",\n    is_flag=True,\n    help=\"Only list upcoming scheduled run start datetimes\",\n)\n@click.option(\n    \"--num-upcoming\",\n    type=int,\n    default=10,\n    help=\"The number of upcoming scheduled run start datetimes to list\",\n)\n@pass_context\n@run_async\nasync def describe_schedule(\n    context: MeltanoCloudCLIContext,\n    only_upcoming: bool,\n    num_upcoming: int,\n) -> None:\n    \"\"\"List Meltano Cloud schedules.\"\"\"\n    schedule = await _get_schedule(\n        config=context.config,\n        deployment_name=context.deployment,\n        schedule_name=context.schedule,\n    )\n\n    if not only_upcoming:\n        click.echo(\n            f\"Deployment name: {schedule['deployment_name']}\\n\"\n            f\"Schedule name:   {schedule['schedule_name']}\\n\"\n            f\"Interval:        {schedule['interval']}\\n\"\n            f\"Enabled:         {schedule['enabled']}\",\n        )\n        if schedule[\"enabled\"]:\n            click.echo(\n                \"\\nApproximate starting date and time (UTC) of \"\n                f\"next {num_upcoming} scheduled runs:\",\n            )\n    if schedule[\"enabled\"]:\n        for dt in _next_n_runs(num_upcoming, schedule[\"interval\"]):\n            click.echo(dt.strftime(\"%Y-%m-%d %H:%M\"))\n", "path": "src/cloud-cli/meltano/cloud/cli/schedule.py"}], "after_files": [{"content": "\"\"\"Meltano Cloud `schedules` command.\"\"\"\n\nfrom __future__ import annotations\n\nimport itertools as it\nimport json\nimport typing as t\nfrom datetime import datetime, timedelta, timezone\nfrom http import HTTPStatus\n\nimport click\nimport tabulate\nfrom croniter import croniter, croniter_range\n\nfrom meltano.cloud.api.client import MeltanoCloudClient, MeltanoCloudError\nfrom meltano.cloud.cli.base import pass_context, run_async, shared_option\n\nif t.TYPE_CHECKING:\n    from meltano.cloud.api.config import MeltanoCloudConfig\n    from meltano.cloud.api.types import CloudProjectSchedule\n    from meltano.cloud.cli.base import MeltanoCloudCLIContext\n\nMAX_PAGE_SIZE = 250\n\n\nclass SchedulesCloudClient(MeltanoCloudClient):\n    \"\"\"A Meltano Cloud client with extensions for schedules.\"\"\"\n\n    async def set_schedule_enabled(\n        self,\n        *,\n        deployment_name: str,\n        schedule_name: str,\n        enabled: bool,\n    ):\n        \"\"\"Use PUT to update the enabled state of a Meltano Cloud project schedule.\n\n        Args:\n            deployment: The name of the deployment the schedule belongs to.\n            schedule: The name of the schedule to enable/disable.\n            enabled: Whether the schedule should be enabled.\n\n        Raises:\n            MeltanoCloudError: The Meltano Cloud API responded with an error.\n        \"\"\"\n        async with self.authenticated():\n            try:\n                await self._json_request(\n                    \"PUT\",\n                    \"/\".join(\n                        (\n                            \"/schedules/v1\",\n                            f\"{self.config.tenant_resource_key}\",\n                            f\"{self.config.internal_project_id}\",\n                            deployment_name,\n                            schedule_name,\n                            \"enabled\",\n                        ),\n                    ),\n                    json=enabled,\n                )\n            except MeltanoCloudError as ex:\n                if ex.response.status == HTTPStatus.NOT_FOUND:\n                    ex.response.reason = (\n                        f\"Unable to find schedule named {schedule_name!r} \"\n                        f\"within a deployment named {deployment_name!r}\"\n                    )\n                    raise MeltanoCloudError(ex.response) from ex\n                raise\n\n    async def get_schedule(\n        self,\n        *,\n        deployment_name: str,\n        schedule_name: str,\n    ):\n        \"\"\"Use GET to get a Meltano Cloud project schedule.\n\n        Args:\n            deployment_name: The name of the deployment the schedule belongs to.\n            schedule_name: The name of the schedule.\n\n        Raises:\n            MeltanoCloudError: The Meltano Cloud API responded with an error.\n        \"\"\"\n        async with self.authenticated():\n            try:\n                return await self._json_request(\n                    \"GET\",\n                    \"/\".join(\n                        (\n                            \"/schedules/v1\",\n                            self.config.tenant_resource_key,\n                            self.config.internal_project_id,\n                            deployment_name,\n                            schedule_name,\n                        ),\n                    ),\n                )\n            except MeltanoCloudError as ex:\n                if ex.response.status == HTTPStatus.NOT_FOUND:\n                    ex.response.reason = (\n                        f\"Unable to find schedule named {schedule_name!r} \"\n                        f\"within a deployment named {deployment_name!r}\"\n                    )\n                    raise MeltanoCloudError(ex.response) from ex\n                raise\n\n    async def get_schedules(\n        self,\n        *,\n        deployment_name: str | None = None,\n        page_size: int | None = None,\n        page_token: str | None = None,\n    ):\n        \"\"\"Use GET to get Meltano Cloud project schedules.\n\n        Args:\n            deployment_name: The name of the deployment the schedule belongs to.\n            page_size: The number of items to request per page.\n            page_token: The page token.\n        \"\"\"\n        params: dict[str, t.Any] = {\n            \"page_size\": page_size,\n            \"page_token\": page_token,\n        }\n\n        path = [\n            \"/schedules/v1\",\n            self.config.tenant_resource_key,\n            self.config.internal_project_id,\n        ]\n        if deployment_name is not None:\n            path.append(deployment_name)\n\n        async with self.authenticated():\n            return await self._json_request(\n                \"GET\",\n                \"/\".join(path),\n                params=self.clean_params(params),\n            )\n\n\nasync def _set_enabled_state(\n    *,\n    config: MeltanoCloudConfig,\n    deployment_name: str | None,\n    schedule_name: str | None,\n    enabled: bool,\n):\n    if schedule_name is None:\n        raise click.UsageError(\"Missing option '--schedule'\")\n    if deployment_name is None:\n        raise click.UsageError(\"Missing option '--deployment'\")\n    async with SchedulesCloudClient(config=config) as client:\n        await client.set_schedule_enabled(\n            deployment_name=deployment_name,\n            schedule_name=schedule_name,\n            enabled=enabled,\n        )\n\n\ndeployment_option = shared_option(\n    \"--deployment\",\n    help=\"The name of the Meltano Cloud deployment the schedule belongs to.\",\n)\n\nschedule_option = shared_option(\n    \"--schedule\",\n    help=\"The name of a schedule within the specified deployment.\",\n)\n\n\n@click.group(\"schedule\")\n@deployment_option\n@schedule_option\ndef schedule_group() -> None:\n    \"\"\"Interact with Meltano Cloud project schedules.\"\"\"\n\n\n@schedule_group.command(\"enable\")\n@deployment_option\n@schedule_option\n@pass_context\n@run_async\nasync def enable(context: MeltanoCloudCLIContext) -> None:\n    \"\"\"Enable a Meltano Cloud schedule.\"\"\"\n    await _set_enabled_state(\n        config=context.config,\n        deployment_name=context.deployment,\n        schedule_name=context.schedule,\n        enabled=True,\n    )\n\n\n@schedule_group.command(\"disable\")\n@deployment_option\n@schedule_option\n@pass_context\n@run_async\nasync def disable(context: MeltanoCloudCLIContext) -> None:\n    \"\"\"Disable a Meltano Cloud schedule.\"\"\"\n    await _set_enabled_state(\n        config=context.config,\n        deployment_name=context.deployment,\n        schedule_name=context.schedule,\n        enabled=False,\n    )\n\n\nasync def _get_schedule(\n    config: MeltanoCloudConfig,\n    deployment_name: str | None,\n    schedule_name: str | None,\n) -> CloudProjectSchedule:\n    if schedule_name is None:\n        raise click.UsageError(\"Missing option '--schedule'\")\n    if deployment_name is None:\n        raise click.UsageError(\"Missing option '--deployment'\")\n    async with SchedulesCloudClient(config=config) as client:\n        return await client.get_schedule(\n            deployment_name=deployment_name,\n            schedule_name=schedule_name,\n        )\n\n\nasync def _get_schedules(\n    config: MeltanoCloudConfig,\n    deployment_name: str | None,\n    limit: int,\n) -> list[CloudProjectSchedule]:\n    page_token = None\n    page_size = min(limit, MAX_PAGE_SIZE)\n    results: list[CloudProjectSchedule] = []\n\n    async with SchedulesCloudClient(config=config) as client:\n        while True:\n            response = await client.get_schedules(\n                deployment_name=deployment_name,\n                page_size=page_size,\n                page_token=page_token,\n            )\n\n            results.extend(response[\"results\"])\n\n            if response[\"pagination\"] and len(results) < limit:\n                page_token = response[\"pagination\"][\"next_page_token\"]\n            else:\n                break\n\n    return results[:limit]\n\n\ndef _next_n_runs(n: int, cron_expr: str) -> tuple[datetime, ...]:\n    now = datetime.now(timezone.utc)\n    return tuple(it.islice(croniter(cron_expr, now, ret_type=datetime), n))\n\n\ntimedelta_year = timedelta(days=365)  # noqa: WPS432\n\n\ndef _approx_daily_freq(\n    cron_expr: str,\n    sample_period: timedelta = timedelta_year,\n    num_digits_precision: int = 1,\n) -> str:\n    now = datetime.now(timezone.utc)\n    num_runs = sum(1 for _ in croniter_range(now, now + sample_period, cron_expr))\n    freq = round(num_runs / sample_period.days, num_digits_precision)\n    return \"< 1\" if freq < 1 else str(freq)\n\n\ndef _process_table_row(\n    schedule: CloudProjectSchedule,\n) -> tuple[str, str, str, str, bool]:\n    return (\n        schedule[\"deployment_name\"],\n        schedule[\"schedule_name\"],\n        schedule[\"interval\"],\n        _approx_daily_freq(schedule[\"interval\"]),\n        schedule[\"enabled\"],\n    )\n\n\ndef _format_schedules_table(\n    schedules: list[CloudProjectSchedule],\n    table_format: str,\n) -> str:\n    \"\"\"Format the schedules as a table.\n\n    Args:\n        schedules: The schedules to format.\n\n    Returns:\n        The formatted schedules.\n    \"\"\"\n    return tabulate.tabulate(\n        [_process_table_row(schedule) for schedule in schedules],\n        headers=(\n            \"Deployment\",\n            \"Schedule\",\n            \"Interval\",\n            \"Runs / Day\",\n            \"Enabled\",\n        ),\n        tablefmt=table_format,\n        floatfmt=\".1f\",\n        # To avoid a tabulate bug (IndexError), only set colalign if there are schedules\n        colalign=(\"left\", \"left\", \"left\", \"right\", \"left\") if schedules else (),\n    )\n\n\nschedule_list_formatters = {\n    \"json\": lambda schedules: json.dumps(schedules, indent=2),\n    \"markdown\": lambda schedules: _format_schedules_table(\n        schedules,\n        table_format=\"github\",\n    ),\n    \"terminal\": lambda schedules: _format_schedules_table(\n        schedules,\n        table_format=\"rounded_outline\",\n    ),\n}\n\n\n@schedule_group.command(\"list\")\n@deployment_option\n@click.option(\n    \"--limit\",\n    required=False,\n    type=int,\n    default=10,\n    help=\"The maximum number of schedules to display.\",\n)\n@click.option(\n    \"--format\",\n    \"output_format\",\n    required=False,\n    default=\"terminal\",\n    type=click.Choice((\"terminal\", \"markdown\", \"json\")),\n    help=\"The output format to use.\",\n)\n@pass_context\n@run_async\nasync def list_schedules(\n    context: MeltanoCloudCLIContext,\n    output_format: str,\n    limit: int,\n) -> None:\n    \"\"\"List Meltano Cloud schedules.\"\"\"\n    click.echo(\n        schedule_list_formatters[output_format](\n            await _get_schedules(\n                config=context.config,\n                deployment_name=context.deployment,\n                limit=limit,\n            ),\n        ),\n    )\n\n\n@schedule_group.command(\"describe\")\n@deployment_option\n@schedule_option\n@click.option(\n    \"--only-upcoming\",\n    is_flag=True,\n    help=\"Only list upcoming scheduled run start datetimes\",\n)\n@click.option(\n    \"--num-upcoming\",\n    type=int,\n    default=10,\n    help=\"The number of upcoming scheduled run start datetimes to list\",\n)\n@pass_context\n@run_async\nasync def describe_schedule(\n    context: MeltanoCloudCLIContext,\n    only_upcoming: bool,\n    num_upcoming: int,\n) -> None:\n    \"\"\"Describe a Meltano Cloud schedules & list upcoming scheduled runs.\"\"\"\n    schedule = await _get_schedule(\n        config=context.config,\n        deployment_name=context.deployment,\n        schedule_name=context.schedule,\n    )\n\n    if not only_upcoming:\n        click.echo(\n            f\"Deployment name: {schedule['deployment_name']}\\n\"\n            f\"Schedule name:   {schedule['schedule_name']}\\n\"\n            f\"Interval:        {schedule['interval']}\\n\"\n            f\"Enabled:         {schedule['enabled']}\",\n        )\n        if schedule[\"enabled\"]:\n            click.echo(\n                \"\\nApproximate starting date and time (UTC) of \"\n                f\"next {num_upcoming} scheduled runs:\",\n            )\n    if schedule[\"enabled\"]:\n        for dt in _next_n_runs(num_upcoming, schedule[\"interval\"]):\n            click.echo(dt.strftime(\"%Y-%m-%d %H:%M\"))\n", "path": "src/cloud-cli/meltano/cloud/cli/schedule.py"}]}