from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Literal
from rossum_api.types import JsonDict
HookAction = Literal[
"user_update", # deprecated in favor of `updated` below
"created",
"updated",
"started",
"confirm",
"changed",
"initialize",
"export",
"received",
"manual",
"scheduled",
"interface",
]
HookType = Literal["webhook", "function", "job"]
HookEvent = Literal[
"annotation_status",
"annotation_content",
"email",
"invocation",
"upload",
]
class HookEventAndAction(str, Enum):
"""Supported list of hook events and actions. Format: "event.action".
See https://rossum.app/api/docs/openapi/api/hook/
We need to define Enum to use it in dataclasses in `list[HookEventAndAction]` for validation.
"""
def __str__(self) -> str:
return self.value
def __repr__(self) -> str:
return self.value
ANNOTATION_STATUS_CHANGED = "annotation_status.changed"
ANNOTATION_CONTENT_INITIALIZE = "annotation_content.initialize"
ANNOTATION_CONTENT_STARTED = "annotation_content.started"
ANNOTATION_CONTENT_USER_UPDATE = "annotation_content.user_update" # deprecated
ANNOTATION_CONTENT_UPDATED = "annotation_content.updated"
ANNOTATION_CONTENT_CONFIRM = "annotation_content.confirm"
ANNOTATION_CONTENT_EXPORT = "annotation_content.export"
UPLOAD_CREATED = "upload.created"
EMAIL_RECEIVED = "email.received"
INVOCATION_SCHEDULED = "invocation.scheduled"
INVOCATION_MANUAL = "invocation.manual"
INVOCATION_INTERFACE = "invocation.interface"
HookExtensionSource = Literal["custom", "rossum_store"]
[docs]
@dataclass
class Hook:
"""Hook is an extension of Rossum that is notified when specific event occurs.
Hook object is used to configure what endpoint or function is executed and when.
For an overview of other extension options see `Extensions <https://rossum.app/api/docs/openapi/guides/extensions/>`_.
Notes
-----
Hooks are notified in parallel if ``run_after`` is not specified.
Attributes
----------
id
ID of the hook.
name
Name of the hook.
url
URL of the hook.
active
If set to ``True`` the hook is notified.
config
Configuration of the hook.
test
Input saved for hook testing purposes, see `Test a hook <https://rossum.app/api/docs/openapi/api/hook/#test-hook>`_.
guide
Description how to use the extension.
read_more_url
URL address leading to more info page.
extension_image_url
URL address of extension picture.
type
type of the hook.
metadata
Client data
queues
List of :class:`~rossum_api.models.queue.Queue` objects that use hook object.
run_after
List of all hooks that has to be executed before running this hook.
events
List of events, when the hook should be notified.
For the list of events see `Webhook events <https://rossum.app/api/docs/openapi/api/hook/>`_.
settings
Specific settings that will be included in the payload when executing the hook.
Field is validated with json schema stored in ``settings_schema`` field.
settings_schema
[BETA] JSON schema for settings field validation.
secrets
JSON schema for ``settings`` field validation.
This is in **BETA**.
extension_source
Import source of the extension.
sideload
List of related objects that should be included in hook request.
For the list of events see `Webhook events <https://rossum.app/api/docs/openapi/api/hook/>`_.
token_owner
URL of a :class:`~rossum_api.models.user.User`.
If present, an API access token is generated for this user and sent to the hook.
Users with organization group admin cannot be set as token_owner.
If ``None``, token is not generated.
token_lifetime_s
Lifetime number of seconds for ``rossum_authorization_token`` (min=0, max=7200).
This setting will ensure the token will be valid after hook response is returned.
If ``None``, default lifetime of 600 is used.
description
Hook description text.
References
----------
https://rossum.app/api/docs/openapi/api/hook/
https://rossum.app/api/docs/openapi/api/hook/#test-hook
https://rossum.app/api/docs/openapi/api/hook/
"""
id: int
name: str
url: str
active: bool
config: JsonDict
test: JsonDict
guide: str | None
read_more_url: str | None
extension_image_url: str | None
type: HookType = "webhook"
metadata: JsonDict = field(default_factory=dict)
queues: list[str] = field(default_factory=list)
run_after: list[str] = field(default_factory=list)
events: list[HookEventAndAction] = field(default_factory=list)
settings: JsonDict = field(default_factory=dict)
settings_schema: JsonDict | None = None
secrets: JsonDict = field(default_factory=dict)
extension_source: HookExtensionSource = "custom"
sideload: list[str] = field(default_factory=list)
token_owner: str | None = None
token_lifetime_s: int | None = None
description: str | None = None
[docs]
@dataclass
class HookRunData:
"""Data class for hook execution logs.
HookRunData captures detailed execution logs and metadata for hook runs within the system.
It provides structured logging for tracking hook lifecycle events, performance metrics, and debugging information.
Attributes
----------
log_level
The severity level of the log entry: "INFO" for successful execution, "ERROR" for failures,
or "WARNING" for non-critical issues.
action
The action that triggered the hook (e.g., ``initialize``, ``changed``).
event
The event type that triggered the hook execution.
request_id
Unique identifier for the HTTP request that initiated this hook execution.
organization_id
The ID of the organization that owns the hook configuration.
hook_id
The unique identifier of the hook configuration being executed.
hook_type
The type/category of the hook (e.g., webhook, email, custom integration).
queue_id
Reference to the queue where the document/annotation is located.
annotation_id
Reference to the specific annotation that triggered or is associated with the hook execution.
email_id
Reference to an email record if the hook involves email processing.
message
Concatenation of messages from hook response, or exception/traceback on errors.
request
Serialized representation of the outgoing HTTP request body sent by the hook (typically JSON).
response
Serialized representation of the HTTP response received from the hook endpoint.
start
ISO 8601 timestamp indicating when the hook execution started.
end
ISO 8601 timestamp indicating when the hook execution completed.
settings
Dictionary containing hook-specific configuration settings and parameters used during execution.
status
Text description of the execution status (e.g., "success", "failed", "timeout").
status_code
HTTP status code returned from the hook endpoint (e.g., 200, 404, 500).
If ``None``, no status code is available.
timestamp
ISO 8601 timestamp of when the hook was triggered.
uuid
Unique identifier for this specific hook execution instance, useful for idempotency and deduplication.
output
Captured log messages or exception/traceback from serverless functions.
If ``None``, no output is captured. Not available for webhooks.
Notes
-----
The retention policy for the logs is set to 7 days.
References
----------
https://rossum.app/api/docs/openapi/api/hook/
https://rossum.app/api/docs/openapi/api/hook/
"""
log_level: Literal["INFO", "ERROR", "WARNING"]
action: HookAction
event: HookEvent
request_id: str
organization_id: int
hook_id: int
hook_type: HookType
queue_id: int | None = None
annotation_id: int | None = None
email_id: int | None = None
message: str = ""
request: str | None = None
response: str | None = None
start: str | None = None
end: str | None = None
settings: JsonDict = field(default_factory=dict)
status: str | None = None
status_code: int | None = None
timestamp: str = ""
uuid: str | None = None
output: str | None = None