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