swh.webhooks.interface module#

swh.webhooks.interface.get_config(config_file: str | None = None) → Dict[str, Any][source]#

Read the configuration file config_file.

If an environment variable SWH_CONFIG_FILENAME is defined, this takes precedence over the config_file parameter.

class swh.webhooks.interface.SvixListResponse(*args, **kwargs)[source]#

Bases: Protocol[SvixData]

data: List[SvixData]#

iterator: Unset | None | str#

done: bool#

swh.webhooks.interface.svix_list(svix_list_request: Callable[[str | None], SvixListResponse[SvixData]]) → Iterator[SvixData][source]#

exception swh.webhooks.interface.SvixHttpError(error_dict: Dict[str, str])[source]#: Bases: Exception

class swh.webhooks.interface.EventType(name: str, description: str, schema: Dict[str, Any])[source]#

Bases: object

Webhook event type definition

An event type is defined by a name, a description and a JSON schema.

name: str#: name of the event type, in the form <group>.<event>

description: str#: description of the event type

schema: Dict[str, Any]#: JSON schema describing the payload sent when the event is triggered

class swh.webhooks.interface.SentEvent(event_type_name: str, endpoint_url: str, channel: str | None, headers: Dict[str, Any], msg_id: str, payload: Dict[str, Any], timestamp: datetime, response: str, response_status_code: int)[source]#

Bases: object

Webhook event delivery attempt definition

event_type_name: str#: The type of sent event

endpoint_url: str#: The URL of the targeted endpoint

channel: str | None#: The channel associated to the endpoint

headers: Dict[str, Any]#: HTTP headers sent with POST request

msg_id: str#: Internal message identifier

payload: Dict[str, Any]#: JSON payload sent as POST request body

timestamp: datetime#: The date the request was sent

response: str#: The response sent by the endpoint

response_status_code: int#: The status code of the sent POST request

class swh.webhooks.interface.Endpoint(url: str, event_type_name: str, channel: str | None = None, metadata: ~typing.Dict[str, ~typing.Any] = <factory>)[source]#

Bases: object

Webhook user endpoint definition

url: str#: URL of the endpoint to receive webhook messages

event_type_name: str#: The type of event the endpoint receives

channel: str | None = None#: Optional channel this endpoint listens to, channels are an extra dimension of filtering messages that is orthogonal to event types

metadata: Dict[str, Any]#: Optional metadata associated to the endpoint

property uid#: Unique identifier for the endpoint

class swh.webhooks.interface.Webhooks(config_file: str | None = None, svix_server_url: str | None = None, svix_auth_token: str | None = None)[source]#

Bases: object

Interface for Software Heritage Webhooks management built on top of the Svix framework (https://docs.svix.com/).

Svix makes sending webhooks easy and reliable by offering webhook sending as a service.

Parameters:

svix_server_url – optional URL of the Svix server, retrieved from configuration if not provided
svix_auth_token – optional bearer token used to perform authenticated requests to the Svix REST API, retrieved from configuration if not provided

event_type_create(event_type: EventType) → None[source]#

Create or update a webhook event type.

Parameters:

event_type – The event type to create or update

Raises:

ValueError – if the event type name is not valid
svix.exceptions.HTTPError – if a request to the Svix REST API fails
jsonschema.exceptions.SchemaError – if the JSON schema of the event type is not valid

event_type_get(event_type_name) → EventType[source]#

Get an active event type by its name.

Parameters:

event_type_name – The name of the event type to retrieve

Raises:

ValueError – if there is no event type with this name
svix.exceptions.HTTPError – if a request to the Svix REST API fails

Returns:

The requested event type.

event_types_list() → List[EventType][source]#

List all registered and active event types.

Raises:: svix.exceptions.HTTPError – if a request to the Svix REST API fails
Returns:: A list of all registered event types.

event_type_delete(event_type_name) → None[source]#

Delete an event type.

The event type is not removed from database but is archived, it is no longer listed and no more events of this type can be sent after this operation. It can be unarchived by creating it again.

Parameters:

event_type_name – The name of the event type to delete

Raises:

ValueError – if there is no event type with this name
svix.exceptions.HTTPError – if a request to the Svix REST API fails

endpoint_create(endpoint: Endpoint, secret: str | None = None) → None[source]#

Create or update an endpoint to receive webhook messages.

Parameters:

endpoint – the endpoint to create
secret – secret used to verify the authenticity of webhook messages, it must match the regular expression ^whsec_[a-zA-Z0-9+/=]{32,100}$ and is automatically generated or rotated if not provided

Raises:

ValueError – if the event type associated to the endpoint does not exist
svix.exceptions.HTTPError – if a request to the Svix REST API fails

endpoints_list(event_type_name: str, channel: str | None = None, ascending_order: bool = False, limit: int | None = None) → Iterator[Endpoint][source]#

List all endpoints receiving messages for a given event type.

Parameters:

event_type_name – the name of the event type to retrieve associated endpoints
channel – optional channel name, only endpoints listening to it are listed if provided, please not that endpoints not listening to any channel receive all events and are always listed
ascending_order – whether to retrieve endpoints in the order they were created
limit – maximum number of endpoints to list

Yields:

Endpoints listening to the event type

Raises:

ValueError – if the event type does not exist
svix.exceptions.HTTPError – if a request to the Svix REST API fails

endpoint_get_secret(endpoint: Endpoint) → str[source]#

Get secret for given endpoint to verify webhook signatures.

Parameters:

endpoint – The endpoint to retrieve the secret

Returns:

The endpoint’s secret.

Raises:

ValueError – if the endpoint does not exist
svix.exceptions.HTTPError – if a request to the Svix REST API fails

endpoint_delete(endpoint: Endpoint) → None[source]#

Delete an endpoint.

Parameters:

endpoint – The endpoint to delete

Raises:

ValueError – if the endpoint does not exist
svix.exceptions.HTTPError – if a request to the Svix REST API fails

event_send(event_type_name: str, payload: Dict[str, Any], channel: str | None = None) → Tuple[str, datetime] | None[source]#

Send an event to registered endpoints.

Parameters:

event_type_name – the name of the event type to send
payload – JSON payload of the event
channel – optional channel name, channels are case-sensitive, and endpoints that are filtering for a specific channel will only get messages sent to that specific channel.

Returns:

Sent message id and timestamp as a tuple or None if no endpoints are listening to the event type.

Raises:

ValueError – if the event type does not exist
jsonschema.exceptions.ValidationError – if the payload does not match the event schema
svix.exceptions.HTTPError – if a request to the Svix REST API fails

sent_events_list_for_endpoint(endpoint: Endpoint, before: datetime | None = None, after: datetime | None = None, limit: int | None = None) → Iterator[SentEvent][source]#

List recent events sent to an endpoint.

Parameters:

endpoint – the endpoint to list sent events
before – list sent events before that timezone aware date if provided
after – list sent events after that timezone aware date if provided

Returns:

list of sent events

Raises:

ValueError – if the endpoint does not exist
svix.exceptions.HTTPError – if a request to the Svix REST API fails

sent_events_list_for_event_type(event_type_name: str, channel: str | None = None, before: datetime | None = None, after: datetime | None = None, limit: int | None = None) → Iterator[SentEvent][source]#

List recent events sent for a specific event type.

Parameters:

event_type_name – the name of the event type to list message attempts
channel – optional channel name, channels are case-sensitive, and endpoints that are filtering for a specific channel will only get messages sent to that specific channel.
before – list sent events before that timezone aware date if provided
after – list sent events after that timezone aware date if provided

Returns:

list of sent events

Raises:

ValueError – if the endpoint does not exist
svix.exceptions.HTTPError – if a request to the Svix REST API fails