Pick a list of parameters from context and cast each of them according to the schemas provided
(params, context, schemas)
| 252 | |
| 253 | |
| 254 | def _cast_params_from(params, context, schemas): |
| 255 | """ |
| 256 | Pick a list of parameters from context and cast each of them according to the schemas provided |
| 257 | """ |
| 258 | result = {} |
| 259 | |
| 260 | # First, cast only explicitly provided live parameters |
| 261 | for name in params: |
| 262 | param_schema = {} |
| 263 | for schema in schemas: |
| 264 | if name in schema: |
| 265 | param_schema = schema[name] |
| 266 | result[name] = _cast(context[name], param_schema) |
| 267 | |
| 268 | # Now, iterate over all parameters, and add any to the live set that satisfy ALL of the |
| 269 | # following criteria: |
| 270 | # |
| 271 | # - Have a default value that is a Jinja template |
| 272 | # - Are using the default value (i.e. not being overwritten by an actual live param) |
| 273 | # |
| 274 | # We do this because the execution API controller first determines live params before |
| 275 | # validating params against the schema. So, we want to make sure that if the default |
| 276 | # value is a template, it is rendered and added to the live params before this validation. |
| 277 | for schema in schemas: |
| 278 | for param_name, param_details in schema.items(): |
| 279 | |
| 280 | # Skip if the parameter have immutable set to true in schema |
| 281 | if param_details.get("immutable"): |
| 282 | continue |
| 283 | |
| 284 | # Skip if the parameter doesn't have a default, or if the |
| 285 | # value in the context is identical to the default |
| 286 | if ( |
| 287 | "default" not in param_details |
| 288 | or param_details.get("default") == context[param_name] |
| 289 | ): |
| 290 | continue |
| 291 | |
| 292 | # Skip if the default value isn't a Jinja expression |
| 293 | if not is_jinja_expression(param_details.get("default")): |
| 294 | continue |
| 295 | |
| 296 | # Skip if the parameter is being overridden |
| 297 | if param_name in params: |
| 298 | continue |
| 299 | |
| 300 | result[param_name] = _cast(context[param_name], param_details) |
| 301 | |
| 302 | return result |
| 303 | |
| 304 | |
| 305 | def render_live_params( |
no test coverage detected