uf.webhooks#

Webhook support for uf.

Provides functionality to trigger HTTP callbacks when functions complete, enabling integration with external services.

Functions

`get_global_webhook_manager`()	Get the global webhook manager.
`set_global_webhook_manager`(manager)	Set the global webhook manager.
`webhook`([on, url, manager])	Decorator to add webhooks to a function.

Classes

`WebhookClient`(url[, headers, timeout, ...])	Client for sending webhooks.
`WebhookEvent`(event_type, func_name, params)	Represents a webhook event.
`WebhookManager`()	Manage webhooks for multiple URLs and events.

class uf.webhooks.WebhookClient(url, headers=None, timeout=10.0, retry_count=3)[source]#

Bases: object

Client for sending webhooks.

Example

>>> client = WebhookClient('https://example.com/webhook')
>>> client.send(WebhookEvent('success', 'my_func', {}, result=42))

send(event, async_send=True)[source]#

Send a webhook event.

Parameters:

event (WebhookEvent) – WebhookEvent to send
async_send (bool) – Whether to send asynchronously

Return type:

bool

Returns:

True if sent successfully (for sync sends)

class uf.webhooks.WebhookEvent(event_type, func_name, params, result=None, error=None)[source]#

Bases: object

Represents a webhook event.

event_type#: Type of event (‘success’, ‘failure’, ‘start’)

func_name#: Name of the function

params#: Function parameters

result#: Function result (if success)

error#: Error message (if failure)

timestamp#: When event occurred

to_dict()[source]#

Convert to dictionary for JSON serialization.

Return type:: dict
Returns:: Dictionary representation

class uf.webhooks.WebhookManager[source]#

Bases: object

Manage webhooks for multiple URLs and events.

Example

>>> manager = WebhookManager()
>>> manager.add_webhook('https://example.com/hook1', events=['success'])
>>> manager.add_webhook('https://example.com/hook2', events=['failure'])
>>> manager.trigger('success', 'my_func', {}, result=42)

add_webhook(url, events=None, headers=None, condition=None)[source]#

Add a webhook.

Parameters:

url (str) – Webhook URL
events (Optional[list[str]]) – List of event types to trigger on (None = all)
headers (Optional[dict]) – Optional HTTP headers
condition (Optional[Callable]) – Optional callable to filter events

Return type:

None

list_webhooks()[source]#

List all registered webhooks.

Return type:: list[dict]
Returns:: List of webhook configurations

remove_webhook(url)[source]#

Remove a webhook by URL.

Parameters:: url (str) – Webhook URL
Return type:: bool
Returns:: True if removed

trigger(event_type, func_name, params, result=None, error=None)[source]#

Trigger webhooks for an event.

Parameters:

event_type (str) – Event type
func_name (str) – Function name
params (dict) – Function parameters
result (Any) – Function result
error (Optional[str]) – Error message

Return type:

int

Returns:

Number of webhooks triggered

uf.webhooks.get_global_webhook_manager()[source]#

Get the global webhook manager.

Return type:: Optional[WebhookManager]

uf.webhooks.set_global_webhook_manager(manager)[source]#

Set the global webhook manager.

Return type:: None

uf.webhooks.webhook(on=None, url=None, manager=None)[source]#

Decorator to add webhooks to a function.

Parameters:

on (Optional[list[str]]) – List of events to trigger on (‘success’, ‘failure’, ‘start’)
url (Optional[str]) – Optional webhook URL
manager (Optional[WebhookManager]) – Optional WebhookManager instance

Returns:

Decorator function

Example

>>> @webhook(on=['success', 'failure'])
... def process_order(order_id: int):
...     # Process order
...     return {'status': 'processed'}