Skip to content

API Reference — Clients

Google Admin

gsm.clients.google_admin

Google Workspace Admin Directory API client.

Wraps admin/directory_v1 for domain operations. Handles 409/duplicate responses as success for idempotency. Retries transient errors (429, 5xx, network timeouts) with exponential backoff.

GoogleAdminError

Bases: RuntimeError

Raised when Admin Directory API returns a non-recoverable error.

Source code in src/gsm/clients/google_admin.py 
29
30
class GoogleAdminError(RuntimeError):
    """Raised when Admin Directory API returns a non-recoverable error."""

GoogleAdminClient

Client for Google Admin SDK Directory API (domains + users).

All API calls are retried up to 3 times on transient errors (429, 5xx, network timeouts) with exponential backoff.

Source code in src/gsm/clients/google_admin.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
class GoogleAdminClient:
    """Client for Google Admin SDK Directory API (domains + users).

    All API calls are retried up to 3 times on transient errors (429, 5xx,
    network timeouts) with exponential backoff.
    """

    _retry_backoff: tuple[float, ...] = _DEFAULT_BACKOFF

    def __init__(self, auth: OAuthDesktopAuth) -> None:
        self._auth = auth
        self._service: Any | None = None

    def _admin(self) -> Any:
        if self._service is None:
            self._service = self._auth.build_admin_service()
        return self._service

    def _exec(self, request: Any) -> Any:
        """Execute a Google API request with retry on transient errors.

        Retries on HttpError 429/500/502/503 and network errors (TimeoutError,
        OSError). Non-retryable errors propagate immediately.
        """
        for attempt in range(_DEFAULT_MAX_ATTEMPTS):
            try:
                return request.execute()
            except HttpError as e:
                status = getattr(getattr(e, "resp", None), "status", None)
                if status not in _RETRYABLE_STATUSES or attempt >= _DEFAULT_MAX_ATTEMPTS - 1:
                    raise
                delay = self._retry_backoff[attempt]
                _log.warning(
                    "google_api_retry",
                    attempt=attempt + 1,
                    status=status,
                    delay=delay,
                )
                _time.sleep(delay)
            except (TimeoutError, OSError) as e:
                if attempt >= _DEFAULT_MAX_ATTEMPTS - 1:
                    raise
                delay = self._retry_backoff[attempt]
                _log.warning(
                    "google_api_retry",
                    attempt=attempt + 1,
                    error=str(e),
                    delay=delay,
                )
                _time.sleep(delay)
        # Unreachable, but satisfies type checker
        msg = "retry loop exited unexpectedly"
        raise RuntimeError(msg)

    # ─── Domain Operations ───────────────────────────────────────────────

    @google_api_call("add domain", duplicate_ok=True)
    def add_domain(self, domain: str) -> bool:
        """Add domain to Workspace. Treats already-exists as success."""
        self._exec(
            self._admin()
            .domains()
            .insert(
                customer="my_customer",
                body={"domainName": domain},
            )
        )
        return True

    @google_api_call("list domains")
    def list_domains(self) -> list[dict[str, Any]]:
        """List all domains registered to the Workspace customer."""
        resp = self._exec(self._admin().domains().list(customer="my_customer"))
        return cast(list[dict[str, Any]], resp.get("domains", []))

    # ─── User Operations ─────────────────────────────────────────────────

    @google_api_call("create user", duplicate_ok=True)
    def create_user(
        self,
        *,
        email: str,
        password: str,
        first_name: str,
        last_name: str,
        change_password_at_next_login: bool = False,
    ) -> bool:
        """Create a Workspace user. Treats already-exists as success."""
        body = {
            "primaryEmail": email,
            "name": {
                "givenName": first_name,
                "familyName": last_name,
            },
            "password": password,
            "changePasswordAtNextLogin": change_password_at_next_login,
        }
        self._exec(self._admin().users().insert(body=body))
        return True

    @google_api_call("update password")
    def update_password(
        self, *, email: str, password: str, change_at_next_login: bool = False
    ) -> bool:
        """Update a user's password. Returns True on success."""
        self._exec(
            self._admin()
            .users()
            .update(
                userKey=email,
                body={
                    "password": password,
                    "changePasswordAtNextLogin": change_at_next_login,
                },
            )
        )
        return True

    @google_api_call("suspend user")
    def suspend_user(self, email: str) -> bool:
        """Suspend a user (block login). Idempotent."""
        self._exec(self._admin().users().update(userKey=email, body={"suspended": True}))
        return True

    @google_api_call("unsuspend user")
    def unsuspend_user(self, email: str) -> bool:
        """Unsuspend a user (re-enable login). Idempotent."""
        self._exec(self._admin().users().update(userKey=email, body={"suspended": False}))
        return True

    @google_api_call("list users")
    def list_users(self, domain: str | None = None) -> list[dict[str, Any]]:
        """List users. Optionally filter by domain."""
        kwargs: dict[str, Any] = {"customer": "my_customer", "maxResults": 500}
        if domain:
            kwargs["domain"] = domain
        users: list[dict[str, Any]] = []
        req = self._admin().users().list(**kwargs)
        while req is not None:
            resp = self._exec(req)
            users.extend(resp.get("users") or [])
            req = self._admin().users().list_next(req, resp)
        return users

    @google_api_call("move user to OU")
    def move_user_to_ou(self, email: str, org_unit_path: str) -> bool:
        """Move user to an Organizational Unit."""
        self._exec(self._admin().users().update(userKey=email, body={"orgUnitPath": org_unit_path}))
        return True

    @google_api_call("delete user", not_found_ok=True)
    def delete_user(self, email: str) -> bool:
        """Delete a user permanently (30-day recovery window in Google)."""
        self._exec(self._admin().users().delete(userKey=email))
        return True

    @google_api_call("update user")
    def update_user(self, email: str, **fields: Any) -> bool:
        """Update user fields (name, phone, department, title, etc)."""
        body: dict[str, Any] = {}
        if "first_name" in fields or "last_name" in fields:
            body["name"] = {}
            if "first_name" in fields:
                body["name"]["givenName"] = fields["first_name"]
            if "last_name" in fields:
                body["name"]["familyName"] = fields["last_name"]
        for key in ("department", "title", "phone"):
            if key in fields:
                if key == "phone":
                    body["phones"] = [{"value": fields[key], "type": "work", "primary": True}]
                else:
                    body["organizations"] = [
                        {**body.get("organizations", [{}])[0], key: fields[key]}
                    ]
        if not body:
            return False
        self._exec(self._admin().users().update(userKey=email, body=body))
        return True

    # ─── Alias Operations ────────────────────────────────────────────────

    @google_api_call("add alias", duplicate_ok=True)
    def add_alias(self, email: str, alias: str) -> bool:
        """Add email alias to a user."""
        self._exec(self._admin().users().aliases().insert(userKey=email, body={"alias": alias}))
        return True

    @google_api_call("list aliases")
    def list_aliases(self, email: str) -> list[str]:
        """List all aliases for a user."""
        resp = self._exec(self._admin().users().aliases().list(userKey=email))
        aliases = resp.get("aliases", [])
        return [a.get("alias", "") for a in aliases if a.get("alias")]

    @google_api_call("remove alias", not_found_ok=True)
    def remove_alias(self, email: str, alias: str) -> bool:
        """Remove email alias from a user."""
        self._exec(self._admin().users().aliases().delete(userKey=email, alias=alias))
        return True

    # ─── License Operations ──────────────────────────────────────────────

    @google_api_call("assign license", duplicate_ok=True)
    def assign_license(self, email: str, sku_id: str, product_id: str) -> bool:
        """Assign a license to a user via Licensing API."""
        from googleapiclient.discovery import build as _build

        licensing = _build("licensing", "v1", credentials=self._auth.get_credentials())
        self._exec(
            licensing.licenseAssignments().insert(
                productId=product_id,
                skuId=sku_id,
                body={"userId": email},
            )
        )
        return True

    # ─── Group Operations ────────────────────────────────────────────────

    @google_api_call("create group", duplicate_ok=True)
    def create_group(self, email: str, name: str | None = None, description: str = "") -> bool:
        """Create a Google Group (mailing list). Idempotent."""
        body: dict[str, Any] = {"email": email}
        if name:
            body["name"] = name
        if description:
            body["description"] = description
        self._exec(self._admin().groups().insert(body=body))
        return True

    @google_api_call("list groups")
    def list_groups(self, domain: str | None = None) -> list[dict[str, Any]]:
        """List groups, optionally filtered by domain."""
        kwargs: dict[str, Any] = {"customer": "my_customer", "maxResults": 200}
        if domain:
            kwargs["domain"] = domain
        groups: list[dict[str, Any]] = []
        req = self._admin().groups().list(**kwargs)
        while req is not None:
            resp = self._exec(req)
            groups.extend(resp.get("groups") or [])
            req = self._admin().groups().list_next(req, resp)
        return groups

    @google_api_call("add group member", duplicate_ok=True)
    def add_group_member(self, group_email: str, member_email: str, role: str = "MEMBER") -> bool:
        """Add member to a group. Idempotent."""
        self._exec(
            self._admin()
            .members()
            .insert(
                groupKey=group_email,
                body={"email": member_email, "role": role},
            )
        )
        return True

    @google_api_call("remove group member", not_found_ok=True)
    def remove_group_member(self, group_email: str, member_email: str) -> bool:
        """Remove member from a group."""
        self._exec(self._admin().members().delete(groupKey=group_email, memberKey=member_email))
        return True

    @google_api_call("list group members")
    def list_group_members(self, group_email: str) -> list[dict[str, Any]]:
        """List members of a group."""
        members: list[dict[str, Any]] = []
        req = self._admin().members().list(groupKey=group_email)
        while req is not None:
            resp = self._exec(req)
            members.extend(resp.get("members") or [])
            req = self._admin().members().list_next(req, resp)
        return members

add_domain(domain)

Add domain to Workspace. Treats already-exists as success.

Source code in src/gsm/clients/google_admin.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@google_api_call("add domain", duplicate_ok=True)
def add_domain(self, domain: str) -> bool:
    """Add domain to Workspace. Treats already-exists as success."""
    self._exec(
        self._admin()
        .domains()
        .insert(
            customer="my_customer",
            body={"domainName": domain},
        )
    )
    return True

list_domains()

List all domains registered to the Workspace customer.

Source code in src/gsm/clients/google_admin.py 
102
103
104
105
106
@google_api_call("list domains")
def list_domains(self) -> list[dict[str, Any]]:
    """List all domains registered to the Workspace customer."""
    resp = self._exec(self._admin().domains().list(customer="my_customer"))
    return cast(list[dict[str, Any]], resp.get("domains", []))

create_user(*, email, password, first_name, last_name, change_password_at_next_login=False)

Create a Workspace user. Treats already-exists as success.

Source code in src/gsm/clients/google_admin.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@google_api_call("create user", duplicate_ok=True)
def create_user(
    self,
    *,
    email: str,
    password: str,
    first_name: str,
    last_name: str,
    change_password_at_next_login: bool = False,
) -> bool:
    """Create a Workspace user. Treats already-exists as success."""
    body = {
        "primaryEmail": email,
        "name": {
            "givenName": first_name,
            "familyName": last_name,
        },
        "password": password,
        "changePasswordAtNextLogin": change_password_at_next_login,
    }
    self._exec(self._admin().users().insert(body=body))
    return True

update_password(*, email, password, change_at_next_login=False)

Update a user's password. Returns True on success.

Source code in src/gsm/clients/google_admin.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@google_api_call("update password")
def update_password(
    self, *, email: str, password: str, change_at_next_login: bool = False
) -> bool:
    """Update a user's password. Returns True on success."""
    self._exec(
        self._admin()
        .users()
        .update(
            userKey=email,
            body={
                "password": password,
                "changePasswordAtNextLogin": change_at_next_login,
            },
        )
    )
    return True

suspend_user(email)

Suspend a user (block login). Idempotent.

Source code in src/gsm/clients/google_admin.py 
151
152
153
154
155
@google_api_call("suspend user")
def suspend_user(self, email: str) -> bool:
    """Suspend a user (block login). Idempotent."""
    self._exec(self._admin().users().update(userKey=email, body={"suspended": True}))
    return True

unsuspend_user(email)

Unsuspend a user (re-enable login). Idempotent.

Source code in src/gsm/clients/google_admin.py 
157
158
159
160
161
@google_api_call("unsuspend user")
def unsuspend_user(self, email: str) -> bool:
    """Unsuspend a user (re-enable login). Idempotent."""
    self._exec(self._admin().users().update(userKey=email, body={"suspended": False}))
    return True

list_users(domain=None)

List users. Optionally filter by domain.

Source code in src/gsm/clients/google_admin.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
@google_api_call("list users")
def list_users(self, domain: str | None = None) -> list[dict[str, Any]]:
    """List users. Optionally filter by domain."""
    kwargs: dict[str, Any] = {"customer": "my_customer", "maxResults": 500}
    if domain:
        kwargs["domain"] = domain
    users: list[dict[str, Any]] = []
    req = self._admin().users().list(**kwargs)
    while req is not None:
        resp = self._exec(req)
        users.extend(resp.get("users") or [])
        req = self._admin().users().list_next(req, resp)
    return users

move_user_to_ou(email, org_unit_path)

Move user to an Organizational Unit.

Source code in src/gsm/clients/google_admin.py 
177
178
179
180
181
@google_api_call("move user to OU")
def move_user_to_ou(self, email: str, org_unit_path: str) -> bool:
    """Move user to an Organizational Unit."""
    self._exec(self._admin().users().update(userKey=email, body={"orgUnitPath": org_unit_path}))
    return True

delete_user(email)

Delete a user permanently (30-day recovery window in Google).

Source code in src/gsm/clients/google_admin.py 
183
184
185
186
187
@google_api_call("delete user", not_found_ok=True)
def delete_user(self, email: str) -> bool:
    """Delete a user permanently (30-day recovery window in Google)."""
    self._exec(self._admin().users().delete(userKey=email))
    return True

update_user(email, **fields)

Update user fields (name, phone, department, title, etc).

Source code in src/gsm/clients/google_admin.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
@google_api_call("update user")
def update_user(self, email: str, **fields: Any) -> bool:
    """Update user fields (name, phone, department, title, etc)."""
    body: dict[str, Any] = {}
    if "first_name" in fields or "last_name" in fields:
        body["name"] = {}
        if "first_name" in fields:
            body["name"]["givenName"] = fields["first_name"]
        if "last_name" in fields:
            body["name"]["familyName"] = fields["last_name"]
    for key in ("department", "title", "phone"):
        if key in fields:
            if key == "phone":
                body["phones"] = [{"value": fields[key], "type": "work", "primary": True}]
            else:
                body["organizations"] = [
                    {**body.get("organizations", [{}])[0], key: fields[key]}
                ]
    if not body:
        return False
    self._exec(self._admin().users().update(userKey=email, body=body))
    return True

add_alias(email, alias)

Add email alias to a user.

Source code in src/gsm/clients/google_admin.py 
214
215
216
217
218
@google_api_call("add alias", duplicate_ok=True)
def add_alias(self, email: str, alias: str) -> bool:
    """Add email alias to a user."""
    self._exec(self._admin().users().aliases().insert(userKey=email, body={"alias": alias}))
    return True

list_aliases(email)

List all aliases for a user.

Source code in src/gsm/clients/google_admin.py 
220
221
222
223
224
225
@google_api_call("list aliases")
def list_aliases(self, email: str) -> list[str]:
    """List all aliases for a user."""
    resp = self._exec(self._admin().users().aliases().list(userKey=email))
    aliases = resp.get("aliases", [])
    return [a.get("alias", "") for a in aliases if a.get("alias")]

remove_alias(email, alias)

Remove email alias from a user.

Source code in src/gsm/clients/google_admin.py 
227
228
229
230
231
@google_api_call("remove alias", not_found_ok=True)
def remove_alias(self, email: str, alias: str) -> bool:
    """Remove email alias from a user."""
    self._exec(self._admin().users().aliases().delete(userKey=email, alias=alias))
    return True

assign_license(email, sku_id, product_id)

Assign a license to a user via Licensing API.

Source code in src/gsm/clients/google_admin.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
@google_api_call("assign license", duplicate_ok=True)
def assign_license(self, email: str, sku_id: str, product_id: str) -> bool:
    """Assign a license to a user via Licensing API."""
    from googleapiclient.discovery import build as _build

    licensing = _build("licensing", "v1", credentials=self._auth.get_credentials())
    self._exec(
        licensing.licenseAssignments().insert(
            productId=product_id,
            skuId=sku_id,
            body={"userId": email},
        )
    )
    return True

create_group(email, name=None, description='')

Create a Google Group (mailing list). Idempotent.

Source code in src/gsm/clients/google_admin.py 
252
253
254
255
256
257
258
259
260
261
@google_api_call("create group", duplicate_ok=True)
def create_group(self, email: str, name: str | None = None, description: str = "") -> bool:
    """Create a Google Group (mailing list). Idempotent."""
    body: dict[str, Any] = {"email": email}
    if name:
        body["name"] = name
    if description:
        body["description"] = description
    self._exec(self._admin().groups().insert(body=body))
    return True

list_groups(domain=None)

List groups, optionally filtered by domain.

Source code in src/gsm/clients/google_admin.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
@google_api_call("list groups")
def list_groups(self, domain: str | None = None) -> list[dict[str, Any]]:
    """List groups, optionally filtered by domain."""
    kwargs: dict[str, Any] = {"customer": "my_customer", "maxResults": 200}
    if domain:
        kwargs["domain"] = domain
    groups: list[dict[str, Any]] = []
    req = self._admin().groups().list(**kwargs)
    while req is not None:
        resp = self._exec(req)
        groups.extend(resp.get("groups") or [])
        req = self._admin().groups().list_next(req, resp)
    return groups

add_group_member(group_email, member_email, role='MEMBER')

Add member to a group. Idempotent.

Source code in src/gsm/clients/google_admin.py 
277
278
279
280
281
282
283
284
285
286
287
288
@google_api_call("add group member", duplicate_ok=True)
def add_group_member(self, group_email: str, member_email: str, role: str = "MEMBER") -> bool:
    """Add member to a group. Idempotent."""
    self._exec(
        self._admin()
        .members()
        .insert(
            groupKey=group_email,
            body={"email": member_email, "role": role},
        )
    )
    return True

remove_group_member(group_email, member_email)

Remove member from a group.

Source code in src/gsm/clients/google_admin.py 
290
291
292
293
294
@google_api_call("remove group member", not_found_ok=True)
def remove_group_member(self, group_email: str, member_email: str) -> bool:
    """Remove member from a group."""
    self._exec(self._admin().members().delete(groupKey=group_email, memberKey=member_email))
    return True

list_group_members(group_email)

List members of a group.

Source code in src/gsm/clients/google_admin.py 
296
297
298
299
300
301
302
303
304
305
@google_api_call("list group members")
def list_group_members(self, group_email: str) -> list[dict[str, Any]]:
    """List members of a group."""
    members: list[dict[str, Any]] = []
    req = self._admin().members().list(groupKey=group_email)
    while req is not None:
        resp = self._exec(req)
        members.extend(resp.get("members") or [])
        req = self._admin().members().list_next(req, resp)
    return members

Cloudflare

gsm.clients.cloudflare

Cloudflare API client - sync requests, idempotent operations.

Ports legacy patterns from gsuite_cloudflare_bot.py: - Zone create: handle code 1061 (already exists) -> GET zone by name - DNS record insert: treat code 81057/81058 (already exists) as success - Bearer token auth, JSON content

CloudflareError

Bases: RuntimeError

Raised when Cloudflare API returns a non-recoverable error.

Source code in src/gsm/clients/cloudflare.py 
24
25
26
27
28
29
class CloudflareError(RuntimeError):
    """Raised when Cloudflare API returns a non-recoverable error."""

    def __init__(self, message: str, *, code: int | None = None) -> None:
        super().__init__(message)
        self.code = code

ZoneInfo dataclass

Result of a zone lookup or creation.

created indicates whether the zone was newly created during this call.

Source code in src/gsm/clients/cloudflare.py 
32
33
34
35
36
37
38
39
40
41
42
@dataclass(frozen=True)
class ZoneInfo:
    """Result of a zone lookup or creation.

    `created` indicates whether the zone was newly created during this call.
    """

    zone_id: str
    name: str
    nameservers: list[str]
    created: bool

CloudflareClient

Thin sync wrapper around Cloudflare v4 API.

Methods are idempotent: existing zones / records are treated as success so workflows can be safely re-run.

Source code in src/gsm/clients/cloudflare.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
class CloudflareClient:
    """Thin sync wrapper around Cloudflare v4 API.

    Methods are idempotent: existing zones / records are treated as success
    so workflows can be safely re-run.
    """

    def __init__(self, settings: Settings) -> None:
        self._settings = settings
        self._token = settings.cf_api_token.get_secret_value()
        self._account_id = settings.cf_account_id
        self._session = requests.Session()
        self._session.headers.update(
            {
                "Authorization": f"Bearer {self._token}",
                "Content-Type": "application/json",
            }
        )

    def _request(
        self,
        method: str,
        url: str,
        *,
        json: Any = None,
        params: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        import time as _time

        max_attempts = 3
        backoff = [1.0, 3.0, 5.0]
        last_exc: Exception | None = None

        for attempt in range(max_attempts):
            try:
                resp = self._session.request(
                    method, url, json=json, params=params, timeout=DEFAULT_TIMEOUT
                )
                if resp.status_code in (502, 503, 429) and attempt < max_attempts - 1:
                    _time.sleep(backoff[attempt])
                    continue
                try:
                    return resp.json()  # type: ignore[no-any-return]
                except ValueError as e:
                    raise CloudflareError(
                        f"non-JSON response from Cloudflare (status={resp.status_code})"
                    ) from e
            except requests.RequestException as e:
                last_exc = e
                if attempt < max_attempts - 1:
                    _time.sleep(backoff[attempt])
                    continue
                raise CloudflareError(
                    f"network error talking to Cloudflare ({method} {url}): {e}"
                ) from e

        raise CloudflareError(f"failed after {max_attempts} attempts ({method} {url}): {last_exc}")

    def ensure_zone(self, domain: str) -> ZoneInfo:
        """Create zone or fetch existing one. Returns ZoneInfo with nameservers populated.

        Raises CloudflareError on hard failures (auth, network, unexpected codes).
        """
        created = self._try_create_zone(domain)
        zone = self._get_zone_by_name(domain)
        if zone is None:
            raise CloudflareError(f"zone not found after ensure: {domain}")
        return ZoneInfo(
            zone_id=zone["id"],
            name=zone["name"],
            nameservers=list(zone.get("name_servers", [])),
            created=created,
        )

    def get_zone_by_name(self, domain: str) -> ZoneInfo | None:
        """Fetch zone info by name. Returns None if zone does not exist."""
        zone = self._get_zone_by_name(domain)
        if zone is None:
            return None
        return ZoneInfo(
            zone_id=zone["id"],
            name=zone["name"],
            nameservers=list(zone.get("name_servers", [])),
            created=False,
        )

    def list_zones(self) -> list[ZoneInfo]:
        """List all zones in the configured CF account (paginated, fetches all)."""
        all_zones: list[ZoneInfo] = []
        page = 1
        while True:
            data = self._request(
                "GET",
                f"{CF_BASE_URL}/zones",
                params={
                    "account.id": self._account_id,
                    "per_page": 50,
                    "page": page,
                },
            )
            if not data.get("success"):
                msg = "; ".join(e.get("message", "?") for e in data.get("errors", []))
                raise CloudflareError(f"failed to list zones: {msg}")
            for zone in data.get("result", []):
                all_zones.append(
                    ZoneInfo(
                        zone_id=zone["id"],
                        name=zone["name"],
                        nameservers=list(zone.get("name_servers", [])),
                        created=False,
                    )
                )
            info = data.get("result_info", {})
            total_pages = info.get("total_pages", 1)
            if page >= total_pages:
                break
            page += 1
        return all_zones

    def get_email_routing_status(self, zone_id: str) -> bool | None:
        """Return True if Email Routing enabled, False if disabled, None if unset."""
        try:
            data = self._request("GET", f"{CF_BASE_URL}/zones/{zone_id}/email/routing")
        except CloudflareError:
            return None
        if not data.get("success"):
            return None
        result = data.get("result") or {}
        if not isinstance(result, dict):
            return None
        return bool(result.get("enabled", False))

    def disable_email_routing(self, zone_id: str) -> bool:
        """Disable Cloudflare Email Routing on a zone (idempotent).

        Returns True if disabled (or was already disabled), raises on hard failure.
        Required before injecting custom MX records (Workspace integration).
        """
        data = self._request("POST", f"{CF_BASE_URL}/zones/{zone_id}/email/routing/disable")
        if data.get("success"):
            return True
        for err in data.get("errors", []):
            msg = err.get("message", "").lower()
            if "not enabled" in msg or "already disabled" in msg or "unconfigured" in msg:
                return True
        msg = "; ".join(e.get("message", "?") for e in data.get("errors", []))
        raise CloudflareError(f"failed to disable email routing on zone {zone_id}: {msg}")

    def upsert_dns_record(
        self,
        zone_id: str,
        *,
        record_type: str,
        name: str,
        content: str,
        priority: int | None = None,
        ttl: int = 1,
        proxied: bool = False,
    ) -> bool:
        """Create DNS record, treating "already exists" responses (81057/81058) as success.

        Returns True when record exists in CF after the call.
        """
        url = f"{CF_BASE_URL}/zones/{zone_id}/dns_records"
        payload: dict[str, Any] = {
            "type": record_type,
            "name": name,
            "content": content,
            "ttl": ttl,
            "proxied": proxied,
        }
        if priority is not None:
            payload["priority"] = priority

        data = self._request("POST", url, json=payload)

        if data.get("success"):
            return True

        for err in data.get("errors", []):
            code = err.get("code", 0)
            msg = err.get("message", "").lower()
            if code in (81057, 81058) or "already exists" in msg:
                return True

        msg = "; ".join(e.get("message", "Unknown") for e in data.get("errors", []))
        raise CloudflareError(
            f"failed to create DNS {record_type} {name}={content}: {msg}",
            code=data.get("errors", [{}])[0].get("code"),
        )

    def _try_create_zone(self, domain: str) -> bool:
        """Attempt to create zone. Returns True if newly created, False if already existed."""
        url = f"{CF_BASE_URL}/zones"
        payload = {
            "name": domain,
            "account": {"id": self._account_id},
            "type": "full",
        }
        data = self._request("POST", url, json=payload)

        if data.get("success"):
            return True

        for err in data.get("errors", []):
            if err.get("code") == 1061:
                return False

        msg = "; ".join(e.get("message", "Unknown") for e in data.get("errors", []))
        raise CloudflareError(
            f"failed to create zone {domain}: {msg}",
            code=data.get("errors", [{}])[0].get("code"),
        )

    def _get_zone_by_name(self, domain: str) -> dict[str, Any] | None:
        url = f"{CF_BASE_URL}/zones"
        data = self._request("GET", url, params={"name": domain})
        if not data.get("success"):
            msg = "; ".join(e.get("message", "Unknown") for e in data.get("errors", []))
            raise CloudflareError(f"failed to query zone {domain}: {msg}")
        results: list[dict[str, Any]] = data.get("result", [])
        if not results:
            return None
        return results[0]

ensure_zone(domain)

Create zone or fetch existing one. Returns ZoneInfo with nameservers populated.

Raises CloudflareError on hard failures (auth, network, unexpected codes).

Source code in src/gsm/clients/cloudflare.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def ensure_zone(self, domain: str) -> ZoneInfo:
    """Create zone or fetch existing one. Returns ZoneInfo with nameservers populated.

    Raises CloudflareError on hard failures (auth, network, unexpected codes).
    """
    created = self._try_create_zone(domain)
    zone = self._get_zone_by_name(domain)
    if zone is None:
        raise CloudflareError(f"zone not found after ensure: {domain}")
    return ZoneInfo(
        zone_id=zone["id"],
        name=zone["name"],
        nameservers=list(zone.get("name_servers", [])),
        created=created,
    )

get_zone_by_name(domain)

Fetch zone info by name. Returns None if zone does not exist.

Source code in src/gsm/clients/cloudflare.py 
119
120
121
122
123
124
125
126
127
128
129
def get_zone_by_name(self, domain: str) -> ZoneInfo | None:
    """Fetch zone info by name. Returns None if zone does not exist."""
    zone = self._get_zone_by_name(domain)
    if zone is None:
        return None
    return ZoneInfo(
        zone_id=zone["id"],
        name=zone["name"],
        nameservers=list(zone.get("name_servers", [])),
        created=False,
    )

list_zones()

List all zones in the configured CF account (paginated, fetches all).

Source code in src/gsm/clients/cloudflare.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def list_zones(self) -> list[ZoneInfo]:
    """List all zones in the configured CF account (paginated, fetches all)."""
    all_zones: list[ZoneInfo] = []
    page = 1
    while True:
        data = self._request(
            "GET",
            f"{CF_BASE_URL}/zones",
            params={
                "account.id": self._account_id,
                "per_page": 50,
                "page": page,
            },
        )
        if not data.get("success"):
            msg = "; ".join(e.get("message", "?") for e in data.get("errors", []))
            raise CloudflareError(f"failed to list zones: {msg}")
        for zone in data.get("result", []):
            all_zones.append(
                ZoneInfo(
                    zone_id=zone["id"],
                    name=zone["name"],
                    nameservers=list(zone.get("name_servers", [])),
                    created=False,
                )
            )
        info = data.get("result_info", {})
        total_pages = info.get("total_pages", 1)
        if page >= total_pages:
            break
        page += 1
    return all_zones

get_email_routing_status(zone_id)

Return True if Email Routing enabled, False if disabled, None if unset.

Source code in src/gsm/clients/cloudflare.py 
164
165
166
167
168
169
170
171
172
173
174
175
def get_email_routing_status(self, zone_id: str) -> bool | None:
    """Return True if Email Routing enabled, False if disabled, None if unset."""
    try:
        data = self._request("GET", f"{CF_BASE_URL}/zones/{zone_id}/email/routing")
    except CloudflareError:
        return None
    if not data.get("success"):
        return None
    result = data.get("result") or {}
    if not isinstance(result, dict):
        return None
    return bool(result.get("enabled", False))

disable_email_routing(zone_id)

Disable Cloudflare Email Routing on a zone (idempotent).

Returns True if disabled (or was already disabled), raises on hard failure. Required before injecting custom MX records (Workspace integration).

Source code in src/gsm/clients/cloudflare.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def disable_email_routing(self, zone_id: str) -> bool:
    """Disable Cloudflare Email Routing on a zone (idempotent).

    Returns True if disabled (or was already disabled), raises on hard failure.
    Required before injecting custom MX records (Workspace integration).
    """
    data = self._request("POST", f"{CF_BASE_URL}/zones/{zone_id}/email/routing/disable")
    if data.get("success"):
        return True
    for err in data.get("errors", []):
        msg = err.get("message", "").lower()
        if "not enabled" in msg or "already disabled" in msg or "unconfigured" in msg:
            return True
    msg = "; ".join(e.get("message", "?") for e in data.get("errors", []))
    raise CloudflareError(f"failed to disable email routing on zone {zone_id}: {msg}")

upsert_dns_record(zone_id, *, record_type, name, content, priority=None, ttl=1, proxied=False)

Create DNS record, treating "already exists" responses (81057/81058) as success.

Returns True when record exists in CF after the call.

Source code in src/gsm/clients/cloudflare.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def upsert_dns_record(
    self,
    zone_id: str,
    *,
    record_type: str,
    name: str,
    content: str,
    priority: int | None = None,
    ttl: int = 1,
    proxied: bool = False,
) -> bool:
    """Create DNS record, treating "already exists" responses (81057/81058) as success.

    Returns True when record exists in CF after the call.
    """
    url = f"{CF_BASE_URL}/zones/{zone_id}/dns_records"
    payload: dict[str, Any] = {
        "type": record_type,
        "name": name,
        "content": content,
        "ttl": ttl,
        "proxied": proxied,
    }
    if priority is not None:
        payload["priority"] = priority

    data = self._request("POST", url, json=payload)

    if data.get("success"):
        return True

    for err in data.get("errors", []):
        code = err.get("code", 0)
        msg = err.get("message", "").lower()
        if code in (81057, 81058) or "already exists" in msg:
            return True

    msg = "; ".join(e.get("message", "Unknown") for e in data.get("errors", []))
    raise CloudflareError(
        f"failed to create DNS {record_type} {name}={content}: {msg}",
        code=data.get("errors", [{}])[0].get("code"),
    )