Skip to content

validation

azure_bootstrap.validation

Lightweight JSON / queue-message schema validation.

Designed for the narrow case of validating untrusted queue payloads cheaply at the consumer entry point — not a pydantic replacement. Schema failures raise :class:InvalidMessageError (unrecoverable per the v2 exception contract) so the Service Bus consumer wrapper dead-letters them.

Classes:

Name Description
InvalidMessageError

Queue payload failed schema validation.

Functions:

Name Description
validate_message

Validate payload against schema. Returns the dict on success.
queue_message_schema

Build a sensible MessageSchema for the common consumer case.

InvalidMessageError

Bases: UnrecoverableError

Queue payload failed schema validation.

validate_message 

validate_message(payload: Any, schema: MessageSchema, *, raise_unrecoverable: bool = True) -> dict[str, Any]

Validate payload against schema. Returns the dict on success.

Failures (non-dict, missing field, type mismatch, pattern, forbidden substring/prefix, required_prefix) bump the namespaced counter and raise :class:InvalidMessageError by default.

Source code in azure_bootstrap/validation/__init__.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def validate_message(
    payload: Any,
    schema: MessageSchema,
    *,
    raise_unrecoverable: bool = True,
) -> dict[str, Any]:
    """Validate ``payload`` against ``schema``. Returns the dict on success.

    Failures (non-dict, missing field, type mismatch, pattern, forbidden
    substring/prefix, required_prefix) bump the namespaced counter and
    raise :class:`InvalidMessageError` by default.
    """
    if not isinstance(payload, dict):
        _bump_rejection(schema)
        if raise_unrecoverable:
            raise InvalidMessageError(
                f"payload is not a JSON object (got {type(payload).__name__})"
            )
        return {}

    for rule in schema.fields:
        reason = _check_field(rule, payload)
        if reason:
            _bump_rejection(schema)
            _logger.warning(
                "validate_message: rejected — %s",
                reason,
                extra={
                    "operation": "validate_message",
                    "schema_namespace": schema.counter_namespace,
                    "reason": reason,
                },
            )
            if raise_unrecoverable:
                raise InvalidMessageError(reason)
            return {}

    return payload

queue_message_schema 

queue_message_schema(*, required_fields: Iterable[str] = ('correlation_id',), path_field: str | None = None, path_required_prefix: str | None = None, counter_namespace: str = 'queue_message') -> MessageSchema

Build a sensible MessageSchema for the common consumer case.

Adds path-traversal defense (forbidden substrings .. and ://) when path_field is supplied. required_prefix enforces a blob / container scoping convention.

Source code in azure_bootstrap/validation/__init__.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def queue_message_schema(
    *,
    required_fields: Iterable[str] = ("correlation_id",),
    path_field: str | None = None,
    path_required_prefix: str | None = None,
    counter_namespace: str = "queue_message",
) -> MessageSchema:
    """Build a sensible MessageSchema for the common consumer case.

    Adds path-traversal defense (forbidden substrings ``..`` and ``://``)
    when ``path_field`` is supplied. ``required_prefix`` enforces a blob /
    container scoping convention.
    """
    rules: list[FieldRule] = [
        FieldRule(name=name, required=True, type=str, non_empty=True) for name in required_fields
    ]
    if path_field is not None:
        rules.append(
            FieldRule(
                name=path_field,
                required=True,
                type=str,
                non_empty=True,
                forbidden_substrings=("..", "://"),
                required_prefix=path_required_prefix,
            )
        )
    return MessageSchema(fields=tuple(rules), counter_namespace=counter_namespace)