Skip to content

Webhooks

Webhooks (V2, beta). Every call sends Affinity API version 2026-07-15 unless pinned newer.

Source code in affinity/services/webhooks.py 
 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
class WebhookService:
    """Webhooks (V2, beta). Every call sends Affinity API version 2026-07-15 unless pinned newer."""

    def __init__(self, client: HTTPClient):
        self._client = client

    def list(
        self,
        *,
        ids: Sequence[WebhookId] | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        updated_before: datetime | None = None,
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[Webhook]:
        """One page of webhooks, ordered by id.

        Without permission to manage all webhooks, only the caller's own webhooks are listed.
        ``created_after``/``updated_after`` are inclusive, ``created_before``/``updated_before``
        exclusive. ``limit`` is 1-100. Pass ``cursor`` (``pagination.next_cursor``) alone for the
        next page; :meth:`iter` follows every page.
        """
        ids = materialize_ids(ids)
        check_cursor_alone(
            cursor,
            ids,
            created_after,
            created_before,
            updated_after,
            updated_before,
            limit,
            total_count,
        )
        if cursor is not None:
            return page_of(
                Webhook, self._client.get_url(cursor, min_api_version=WEBHOOKS_MIN_API_VERSION)
            )
        params = webhook_params(
            ids=ids,
            created_after=created_after,
            created_before=created_before,
            updated_after=updated_after,
            updated_before=updated_before,
            limit=limit,
            total_count=total_count,
        )
        if no_ids(ids):
            return empty_page(Webhook, total_count=total_count)
        return page_of(
            Webhook,
            self._client.get(
                "/webhooks", params=params or None, min_api_version=WEBHOOKS_MIN_API_VERSION
            ),
        )

    def iter(
        self,
        *,
        ids: Sequence[WebhookId] | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        updated_before: datetime | None = None,
        limit: int | None = None,
    ) -> Iterator[Webhook]:
        """All webhooks matching the filters (every page); see :meth:`list`."""

        def fetch_page(next_url: str | None) -> PaginatedResponse[Webhook]:
            if next_url:
                return self.list(cursor=next_url)
            return self.list(
                ids=ids,
                created_after=created_after,
                created_before=created_before,
                updated_after=updated_after,
                updated_before=updated_before,
                limit=limit,
            )

        return PageIterator(fetch_page)

    def get(self, webhook_id: WebhookId) -> Webhook:
        """One webhook. ``NotFoundError`` when it does not exist or is not visible to the caller."""
        return Webhook.model_validate(
            self._client.get(
                f"/webhooks/{int(webhook_id)}", min_api_version=WEBHOOKS_MIN_API_VERSION
            )
        )

    def create(self, data: WebhookCreate) -> Webhook:
        """Create a webhook; the caller becomes its creator.

        Affinity first sends a test delivery to ``data.url``. A URL that is unreachable, holds
        credentials, or is already used by another webhook raises ``ValidationError``; reaching
        the organization's webhook limit raises one too. ``subscriptions=[]`` = every event type.
        """
        return Webhook.model_validate(
            self._client.post(
                "/webhooks", json=create_body(data), min_api_version=WEBHOOKS_MIN_API_VERSION
            )
        )

    def update(self, webhook_id: WebhookId, data: WebhookUpdate) -> Webhook:
        """Change a webhook (only the fields set in ``data``) and return it as read back.

        Two requests: the update answers with no body, so the webhook is fetched afterwards.
        Changing ``url`` on an active webhook, or setting ``status`` to active, sends a test
        delivery first. A webhook you didn't create, without permission to manage all webhooks,
        is not visible to you (``NotFoundError``).
        """
        self._client.post(
            f"/webhooks/{int(webhook_id)}",
            json=update_body(data),
            min_api_version=WEBHOOKS_MIN_API_VERSION,
        )
        return self.get(webhook_id)

    def delete(self, webhook_id: WebhookId) -> None:
        """Delete a webhook. One you didn't create, without permission to manage all webhooks, is
        not visible to you (``NotFoundError``)."""
        self._client.delete(
            f"/webhooks/{int(webhook_id)}", min_api_version=WEBHOOKS_MIN_API_VERSION
        )

create(data: WebhookCreate) -> Webhook

Create a webhook; the caller becomes its creator.

Affinity first sends a test delivery to data.url. A URL that is unreachable, holds credentials, or is already used by another webhook raises ValidationError; reaching the organization's webhook limit raises one too. subscriptions=[] = every event type.

Source code in affinity/services/webhooks.py 
172
173
174
175
176
177
178
179
180
181
182
183
def create(self, data: WebhookCreate) -> Webhook:
    """Create a webhook; the caller becomes its creator.

    Affinity first sends a test delivery to ``data.url``. A URL that is unreachable, holds
    credentials, or is already used by another webhook raises ``ValidationError``; reaching
    the organization's webhook limit raises one too. ``subscriptions=[]`` = every event type.
    """
    return Webhook.model_validate(
        self._client.post(
            "/webhooks", json=create_body(data), min_api_version=WEBHOOKS_MIN_API_VERSION
        )
    )

delete(webhook_id: WebhookId) -> None

Delete a webhook. One you didn't create, without permission to manage all webhooks, is not visible to you (NotFoundError).

Source code in affinity/services/webhooks.py 
200
201
202
203
204
205
def delete(self, webhook_id: WebhookId) -> None:
    """Delete a webhook. One you didn't create, without permission to manage all webhooks, is
    not visible to you (``NotFoundError``)."""
    self._client.delete(
        f"/webhooks/{int(webhook_id)}", min_api_version=WEBHOOKS_MIN_API_VERSION
    )

get(webhook_id: WebhookId) -> Webhook

One webhook. NotFoundError when it does not exist or is not visible to the caller.

Source code in affinity/services/webhooks.py 
164
165
166
167
168
169
170
def get(self, webhook_id: WebhookId) -> Webhook:
    """One webhook. ``NotFoundError`` when it does not exist or is not visible to the caller."""
    return Webhook.model_validate(
        self._client.get(
            f"/webhooks/{int(webhook_id)}", min_api_version=WEBHOOKS_MIN_API_VERSION
        )
    )

iter(*, ids: Sequence[WebhookId] | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, updated_before: datetime | None = None, limit: int | None = None) -> Iterator[Webhook]

All webhooks matching the filters (every page); see :meth:list.

Source code in affinity/services/webhooks.py 
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 iter(
    self,
    *,
    ids: Sequence[WebhookId] | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    updated_before: datetime | None = None,
    limit: int | None = None,
) -> Iterator[Webhook]:
    """All webhooks matching the filters (every page); see :meth:`list`."""

    def fetch_page(next_url: str | None) -> PaginatedResponse[Webhook]:
        if next_url:
            return self.list(cursor=next_url)
        return self.list(
            ids=ids,
            created_after=created_after,
            created_before=created_before,
            updated_after=updated_after,
            updated_before=updated_before,
            limit=limit,
        )

    return PageIterator(fetch_page)

list(*, ids: Sequence[WebhookId] | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, updated_before: datetime | None = None, limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[Webhook]

One page of webhooks, ordered by id.

Without permission to manage all webhooks, only the caller's own webhooks are listed. created_after/updated_after are inclusive, created_before/updated_before exclusive. limit is 1-100. Pass cursor (pagination.next_cursor) alone for the next page; :meth:iter follows every page.

Source code in affinity/services/webhooks.py 
 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
def list(
    self,
    *,
    ids: Sequence[WebhookId] | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    updated_before: datetime | None = None,
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[Webhook]:
    """One page of webhooks, ordered by id.

    Without permission to manage all webhooks, only the caller's own webhooks are listed.
    ``created_after``/``updated_after`` are inclusive, ``created_before``/``updated_before``
    exclusive. ``limit`` is 1-100. Pass ``cursor`` (``pagination.next_cursor``) alone for the
    next page; :meth:`iter` follows every page.
    """
    ids = materialize_ids(ids)
    check_cursor_alone(
        cursor,
        ids,
        created_after,
        created_before,
        updated_after,
        updated_before,
        limit,
        total_count,
    )
    if cursor is not None:
        return page_of(
            Webhook, self._client.get_url(cursor, min_api_version=WEBHOOKS_MIN_API_VERSION)
        )
    params = webhook_params(
        ids=ids,
        created_after=created_after,
        created_before=created_before,
        updated_after=updated_after,
        updated_before=updated_before,
        limit=limit,
        total_count=total_count,
    )
    if no_ids(ids):
        return empty_page(Webhook, total_count=total_count)
    return page_of(
        Webhook,
        self._client.get(
            "/webhooks", params=params or None, min_api_version=WEBHOOKS_MIN_API_VERSION
        ),
    )

update(webhook_id: WebhookId, data: WebhookUpdate) -> Webhook

Change a webhook (only the fields set in data) and return it as read back.

Two requests: the update answers with no body, so the webhook is fetched afterwards. Changing url on an active webhook, or setting status to active, sends a test delivery first. A webhook you didn't create, without permission to manage all webhooks, is not visible to you (NotFoundError).

Source code in affinity/services/webhooks.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def update(self, webhook_id: WebhookId, data: WebhookUpdate) -> Webhook:
    """Change a webhook (only the fields set in ``data``) and return it as read back.

    Two requests: the update answers with no body, so the webhook is fetched afterwards.
    Changing ``url`` on an active webhook, or setting ``status`` to active, sends a test
    delivery first. A webhook you didn't create, without permission to manage all webhooks,
    is not visible to you (``NotFoundError``).
    """
    self._client.post(
        f"/webhooks/{int(webhook_id)}",
        json=update_body(data),
        min_api_version=WEBHOOKS_MIN_API_VERSION,
    )
    return self.get(webhook_id)

Async webhooks (V2, beta); same behavior as :class:WebhookService.

Source code in affinity/services/webhooks.py 
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
306
307
308
309
310
311
312
313
314
315
316
317
class AsyncWebhookService:
    """Async webhooks (V2, beta); same behavior as :class:`WebhookService`."""

    def __init__(self, client: AsyncHTTPClient):
        self._client = client

    async def list(
        self,
        *,
        ids: Sequence[WebhookId] | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        updated_before: datetime | None = None,
        limit: int | None = None,
        total_count: bool = False,
        cursor: str | None = None,
    ) -> PaginatedResponse[Webhook]:
        """One page of webhooks; see :meth:`WebhookService.list`."""
        ids = materialize_ids(ids)
        check_cursor_alone(
            cursor,
            ids,
            created_after,
            created_before,
            updated_after,
            updated_before,
            limit,
            total_count,
        )
        if cursor is not None:
            return page_of(
                Webhook,
                await self._client.get_url(cursor, min_api_version=WEBHOOKS_MIN_API_VERSION),
            )
        params = webhook_params(
            ids=ids,
            created_after=created_after,
            created_before=created_before,
            updated_after=updated_after,
            updated_before=updated_before,
            limit=limit,
            total_count=total_count,
        )
        if no_ids(ids):
            return empty_page(Webhook, total_count=total_count)
        return page_of(
            Webhook,
            await self._client.get(
                "/webhooks", params=params or None, min_api_version=WEBHOOKS_MIN_API_VERSION
            ),
        )

    def iter(
        self,
        *,
        ids: Sequence[WebhookId] | None = None,
        created_after: datetime | None = None,
        created_before: datetime | None = None,
        updated_after: datetime | None = None,
        updated_before: datetime | None = None,
        limit: int | None = None,
    ) -> AsyncIterator[Webhook]:
        """All webhooks matching the filters (every page); see :meth:`list`."""

        async def fetch_page(next_url: str | None) -> PaginatedResponse[Webhook]:
            if next_url:
                return await self.list(cursor=next_url)
            return await self.list(
                ids=ids,
                created_after=created_after,
                created_before=created_before,
                updated_after=updated_after,
                updated_before=updated_before,
                limit=limit,
            )

        return AsyncPageIterator(fetch_page)

    async def get(self, webhook_id: WebhookId) -> Webhook:
        """One webhook; see :meth:`WebhookService.get`."""
        return Webhook.model_validate(
            await self._client.get(
                f"/webhooks/{int(webhook_id)}", min_api_version=WEBHOOKS_MIN_API_VERSION
            )
        )

    async def create(self, data: WebhookCreate) -> Webhook:
        """Create a webhook; see :meth:`WebhookService.create`."""
        return Webhook.model_validate(
            await self._client.post(
                "/webhooks", json=create_body(data), min_api_version=WEBHOOKS_MIN_API_VERSION
            )
        )

    async def update(self, webhook_id: WebhookId, data: WebhookUpdate) -> Webhook:
        """Change a webhook and return it as read back (two requests); see
        :meth:`WebhookService.update`."""
        await self._client.post(
            f"/webhooks/{int(webhook_id)}",
            json=update_body(data),
            min_api_version=WEBHOOKS_MIN_API_VERSION,
        )
        return await self.get(webhook_id)

    async def delete(self, webhook_id: WebhookId) -> None:
        """Delete a webhook; see :meth:`WebhookService.delete`."""
        await self._client.delete(
            f"/webhooks/{int(webhook_id)}", min_api_version=WEBHOOKS_MIN_API_VERSION
        )

create(data: WebhookCreate) -> Webhook async

Create a webhook; see :meth:WebhookService.create.

Source code in affinity/services/webhooks.py 
295
296
297
298
299
300
301
async def create(self, data: WebhookCreate) -> Webhook:
    """Create a webhook; see :meth:`WebhookService.create`."""
    return Webhook.model_validate(
        await self._client.post(
            "/webhooks", json=create_body(data), min_api_version=WEBHOOKS_MIN_API_VERSION
        )
    )

delete(webhook_id: WebhookId) -> None async

Delete a webhook; see :meth:WebhookService.delete.

Source code in affinity/services/webhooks.py 
313
314
315
316
317
async def delete(self, webhook_id: WebhookId) -> None:
    """Delete a webhook; see :meth:`WebhookService.delete`."""
    await self._client.delete(
        f"/webhooks/{int(webhook_id)}", min_api_version=WEBHOOKS_MIN_API_VERSION
    )

get(webhook_id: WebhookId) -> Webhook async

One webhook; see :meth:WebhookService.get.

Source code in affinity/services/webhooks.py 
287
288
289
290
291
292
293
async def get(self, webhook_id: WebhookId) -> Webhook:
    """One webhook; see :meth:`WebhookService.get`."""
    return Webhook.model_validate(
        await self._client.get(
            f"/webhooks/{int(webhook_id)}", min_api_version=WEBHOOKS_MIN_API_VERSION
        )
    )

iter(*, ids: Sequence[WebhookId] | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, updated_before: datetime | None = None, limit: int | None = None) -> AsyncIterator[Webhook]

All webhooks matching the filters (every page); see :meth:list.

Source code in affinity/services/webhooks.py 
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
def iter(
    self,
    *,
    ids: Sequence[WebhookId] | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    updated_before: datetime | None = None,
    limit: int | None = None,
) -> AsyncIterator[Webhook]:
    """All webhooks matching the filters (every page); see :meth:`list`."""

    async def fetch_page(next_url: str | None) -> PaginatedResponse[Webhook]:
        if next_url:
            return await self.list(cursor=next_url)
        return await self.list(
            ids=ids,
            created_after=created_after,
            created_before=created_before,
            updated_after=updated_after,
            updated_before=updated_before,
            limit=limit,
        )

    return AsyncPageIterator(fetch_page)

list(*, ids: Sequence[WebhookId] | None = None, created_after: datetime | None = None, created_before: datetime | None = None, updated_after: datetime | None = None, updated_before: datetime | None = None, limit: int | None = None, total_count: bool = False, cursor: str | None = None) -> PaginatedResponse[Webhook] async

One page of webhooks; see :meth:WebhookService.list.

Source code in affinity/services/webhooks.py 
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
async def list(
    self,
    *,
    ids: Sequence[WebhookId] | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
    updated_after: datetime | None = None,
    updated_before: datetime | None = None,
    limit: int | None = None,
    total_count: bool = False,
    cursor: str | None = None,
) -> PaginatedResponse[Webhook]:
    """One page of webhooks; see :meth:`WebhookService.list`."""
    ids = materialize_ids(ids)
    check_cursor_alone(
        cursor,
        ids,
        created_after,
        created_before,
        updated_after,
        updated_before,
        limit,
        total_count,
    )
    if cursor is not None:
        return page_of(
            Webhook,
            await self._client.get_url(cursor, min_api_version=WEBHOOKS_MIN_API_VERSION),
        )
    params = webhook_params(
        ids=ids,
        created_after=created_after,
        created_before=created_before,
        updated_after=updated_after,
        updated_before=updated_before,
        limit=limit,
        total_count=total_count,
    )
    if no_ids(ids):
        return empty_page(Webhook, total_count=total_count)
    return page_of(
        Webhook,
        await self._client.get(
            "/webhooks", params=params or None, min_api_version=WEBHOOKS_MIN_API_VERSION
        ),
    )

update(webhook_id: WebhookId, data: WebhookUpdate) -> Webhook async

Change a webhook and return it as read back (two requests); see :meth:WebhookService.update.

Source code in affinity/services/webhooks.py 
303
304
305
306
307
308
309
310
311
async def update(self, webhook_id: WebhookId, data: WebhookUpdate) -> Webhook:
    """Change a webhook and return it as read back (two requests); see
    :meth:`WebhookService.update`."""
    await self._client.post(
        f"/webhooks/{int(webhook_id)}",
        json=update_body(data),
        min_api_version=WEBHOOKS_MIN_API_VERSION,
    )
    return await self.get(webhook_id)