This function validates trigger payload parameters for system and user-defined triggers. :param trigger_type_ref: Reference of a trigger type / trigger / trigger dictionary object. :type trigger_type_ref: ``str`` :param payload: Trigger payload. :type payload: ``dict`` :r
(
trigger_type_ref, payload, throw_on_inexistent_trigger=False
)
| 126 | |
| 127 | |
| 128 | def validate_trigger_payload( |
| 129 | trigger_type_ref, payload, throw_on_inexistent_trigger=False |
| 130 | ): |
| 131 | """ |
| 132 | This function validates trigger payload parameters for system and user-defined triggers. |
| 133 | |
| 134 | :param trigger_type_ref: Reference of a trigger type / trigger / trigger dictionary object. |
| 135 | :type trigger_type_ref: ``str`` |
| 136 | |
| 137 | :param payload: Trigger payload. |
| 138 | :type payload: ``dict`` |
| 139 | |
| 140 | :return: Cleaned payload on success, None if validation is not performed. |
| 141 | """ |
| 142 | if not trigger_type_ref: |
| 143 | return None |
| 144 | |
| 145 | # NOTE: Due to the awful code in some other places we also need to support a scenario where |
| 146 | # this variable is a dictionary and contains various TriggerDB object attributes. |
| 147 | if isinstance(trigger_type_ref, dict): |
| 148 | if trigger_type_ref.get("type", None): |
| 149 | trigger_type_ref = trigger_type_ref["type"] |
| 150 | else: |
| 151 | trigger_db = triggers.get_trigger_db_by_ref_or_dict(trigger_type_ref) |
| 152 | |
| 153 | if not trigger_db: |
| 154 | # Corresponding TriggerDB not found, likely a corrupted database, skip the |
| 155 | # validation. |
| 156 | return None |
| 157 | |
| 158 | trigger_type_ref = trigger_db.type |
| 159 | |
| 160 | is_system_trigger = trigger_type_ref in SYSTEM_TRIGGER_TYPES |
| 161 | if is_system_trigger: |
| 162 | # System trigger |
| 163 | payload_schema = SYSTEM_TRIGGER_TYPES[trigger_type_ref]["payload_schema"] |
| 164 | else: |
| 165 | # We assume Trigger ref and not TriggerType ref is passed in if second |
| 166 | # part (trigger name) is a valid UUID version 4 |
| 167 | try: |
| 168 | trigger_uuid = uuid.UUID(trigger_type_ref.split(".")[-1]) |
| 169 | except ValueError: |
| 170 | is_trigger_db = False |
| 171 | else: |
| 172 | is_trigger_db = trigger_uuid.version == 4 |
| 173 | |
| 174 | if is_trigger_db: |
| 175 | trigger_db = triggers.get_trigger_db_by_ref(trigger_type_ref) |
| 176 | |
| 177 | if trigger_db: |
| 178 | trigger_type_ref = trigger_db.type |
| 179 | |
| 180 | trigger_type_db = triggers.get_trigger_type_db(trigger_type_ref) |
| 181 | |
| 182 | if not trigger_type_db: |
| 183 | # Trigger doesn't exist in the database |
| 184 | if throw_on_inexistent_trigger: |
| 185 | msg = ( |
no test coverage detected