Skip to content

prefect.blocks.webhook

Webhook

Bases: Block

Block that enables calling webhooks.

Source code in src/prefect/blocks/webhook.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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
class Webhook(Block):
    """
    Block that enables calling webhooks.
    """

    _block_type_name = "Webhook"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/c7247cb359eb6cf276734d4b1fbf00fb8930e89e-250x250.png"  # type: ignore
    _documentation_url = (
        "https://docs.prefect.io/latest/automate/events/webhook-triggers"
    )

    method: Literal["GET", "POST", "PUT", "PATCH", "DELETE"] = Field(
        default="POST", description="The webhook request method. Defaults to `POST`."
    )

    url: SecretStr = Field(
        default=...,
        title="Webhook URL",
        description="The webhook URL.",
        examples=["https://hooks.slack.com/XXX"],
    )

    headers: SecretDict = Field(
        default_factory=lambda: SecretDict(dict()),
        title="Webhook Headers",
        description="A dictionary of headers to send with the webhook request.",
    )
    allow_private_urls: bool = Field(
        default=True,
        description="Whether to allow notifications to private URLs. Defaults to True.",
    )
    verify: bool = Field(
        default=True,
        description="Whether or not to enforce a secure connection to the webhook.",
    )

    def block_initialization(self):
        if self.verify:
            self._client = AsyncClient(transport=_http_transport)
        else:
            self._client = AsyncClient(transport=_insecure_http_transport)

    async def call(self, payload: Optional[dict] = None) -> Response:
        """
        Call the webhook.

        Args:
            payload: an optional payload to send when calling the webhook.
        """
        if not self.allow_private_urls:
            validate_restricted_url(self.url.get_secret_value())

        async with self._client:
            return await self._client.request(
                method=self.method,
                url=self.url.get_secret_value(),
                headers=self.headers.get_secret_value(),
                json=payload,
            )

call(payload=None) async

Call the webhook.

Parameters:

Name Type Description Default
payload Optional[dict]

an optional payload to send when calling the webhook.

 None
Source code in src/prefect/blocks/webhook.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
async def call(self, payload: Optional[dict] = None) -> Response:
    """
    Call the webhook.

    Args:
        payload: an optional payload to send when calling the webhook.
    """
    if not self.allow_private_urls:
        validate_restricted_url(self.url.get_secret_value())

    async with self._client:
        return await self._client.request(
            method=self.method,
            url=self.url.get_secret_value(),
            headers=self.headers.get_secret_value(),
            json=payload,
        )