feat(freeze): add scheduled freeze CLI commands

README.md

@@ -17,6 +17,11 @@ Upload and analyze CI results to get better visibility into your CI pipeline.
 Learn more about [CI Insights in the
 documentation](https://docs.mergify.com/ci-insights/).
 
+### Scheduled Freezes
+
+Create and manage scheduled freezes to temporarily halt merging of pull requests
+matching specific conditions.
+
 ## Installation
 
 ```shell
@@ -34,6 +39,9 @@ mergify stack --help
 
 # Upload CI results
 mergify ci --help
+
+# Manage scheduled freezes
+mergify freeze --help
 ```
 
 ## Contributing

mergify_cli/cli.py

@@ -23,6 +23,7 @@
 
 from mergify_cli import VERSION
 from mergify_cli.ci import cli as ci_cli_mod
+from mergify_cli.freeze import cli as freeze_cli_mod
 from mergify_cli.stack import cli as stack_cli_mod
 
 
@@ -44,6 +45,7 @@ def cli(
 
 cli.add_command(stack_cli_mod.stack)
 cli.add_command(ci_cli_mod.ci)
+cli.add_command(freeze_cli_mod.freeze)
 
 
 def main() -> None:

mergify_cli/freeze/api.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import typing
+
+
+if typing.TYPE_CHECKING:
+    import datetime
+    import uuid
+
+    import httpx
+
+
+class ScheduledFreezeResponse(typing.TypedDict, total=False):
+    id: typing.Required[str]
+    reason: typing.Required[str]
+    start: typing.Required[str]
+    end: typing.Required[str | None]
+    timezone: typing.Required[str]
+    matching_conditions: typing.Required[list[str]]
+    exclude_conditions: list[str]
+
+
+class CreateScheduledFreezePayload(typing.TypedDict, total=False):
+    reason: typing.Required[str]
+    start: typing.Required[str | None]
+    end: typing.Required[str | None]
+    timezone: typing.Required[str]
+    matching_conditions: typing.Required[list[str]]
+    exclude_conditions: list[str]
+
+
+class UpdateScheduledFreezePayload(typing.TypedDict, total=False):
+    reason: str | None
+    start: str | None
+    end: str | None
+    timezone: str | None
+    matching_conditions: list[str] | None
+    exclude_conditions: list[str] | None
+
+
+async def list_freezes(
+    client: httpx.AsyncClient,
+    repository: str,
+) -> list[ScheduledFreezeResponse]:
+    response = await client.get(
+        f"/v1/repos/{repository}/scheduled_freeze",
+    )
+    data: dict[str, list[ScheduledFreezeResponse]] = response.json()
+    return data["scheduled_freezes"]
+
+
+async def create_freeze(
+    client: httpx.AsyncClient,
+    repository: str,
+    *,
+    reason: str,
+    timezone: str,
+    matching_conditions: list[str],
+    start: datetime.datetime | None = None,
+    end: datetime.datetime | None = None,
+    exclude_conditions: list[str] | None = None,
+) -> ScheduledFreezeResponse:
+    payload: CreateScheduledFreezePayload = {
+        "reason": reason,
+        "start": start.isoformat() if start is not None else None,
+        "end": end.isoformat() if end is not None else None,
+        "timezone": timezone,
+        "matching_conditions": matching_conditions,
+    }
+    if exclude_conditions:
+        payload["exclude_conditions"] = exclude_conditions
+
+    response = await client.post(
+        f"/v1/repos/{repository}/scheduled_freeze",
+        json=payload,
+    )
+    return response.json()  # type: ignore[no-any-return]
+
+
+async def update_freeze(
+    client: httpx.AsyncClient,
+    repository: str,
+    freeze_id: uuid.UUID,
+    *,
+    reason: str | None = None,
+    timezone: str | None = None,
+    matching_conditions: list[str] | None = None,
+    start: datetime.datetime | None = None,
+    end: datetime.datetime | None = None,
+    exclude_conditions: list[str] | None = None,
+) -> ScheduledFreezeResponse:
+    payload: UpdateScheduledFreezePayload = {}
+    if reason is not None:
+        payload["reason"] = reason
+    if timezone is not None:
+        payload["timezone"] = timezone
+    if matching_conditions is not None:
+        payload["matching_conditions"] = matching_conditions
+    if start is not None:
+        payload["start"] = start.isoformat()
+    if end is not None:
+        payload["end"] = end.isoformat()
+    if exclude_conditions is not None:
+        payload["exclude_conditions"] = exclude_conditions
+
+    response = await client.patch(
+        f"/v1/repos/{repository}/scheduled_freeze/{freeze_id}",
+        json=payload,
+    )
+    return response.json()  # type: ignore[no-any-return]
+
+
+async def delete_freeze(
+    client: httpx.AsyncClient,
+    repository: str,
+    freeze_id: uuid.UUID,
+    *,
+    delete_reason: str | None = None,
+) -> None:
+    url = f"/v1/repos/{repository}/scheduled_freeze/{freeze_id}"
+    if delete_reason is not None:
+        await client.request(
+            "DELETE",
+            url,
+            json={"delete_reason": delete_reason},
+        )
+    else:
+        await client.delete(url)