app.engine.dsl_strategy module

BaseStrategy adapter that runs a persisted DSL StrategyDefinition.

This adapter is what StrategyService instantiates for custom: ids; it is also exercised directly by the simulate endpoint and by tests.

Notable choice: _generate_hash_of_calculate_points is overridden to hash the canonicalized AST (sorted JSON keys) instead of the Python source of the method. The built-in strategies still use the inspect-based hash inherited from BaseStrategy, so existing UserPoints idempotency keys remain valid - only DSL strategies opt into the new hash scheme.

class app.engine.dsl_strategy.DslStrategy(definition, interpreter, analytics_service, *, parent_strategy=None, observer=None)[source]

Bases: BaseStrategy

Parameters:

  • definition (StrategyDefinitionRead)

  • interpreter (DslInterpreter)

  • analytics_service (Any)

  • parent_strategy (Optional[BaseStrategy])

  • observer (Optional[Any])

get_strategy_id()[source]

Return the public id for this custom strategy.

Returns:

str – The id in the form "custom:<definition id>".

Return type:

str

async calculate_points(externalGameId=None, externalTaskId=None, externalUserId=None, data=None)[source]

Score an event by running this custom strategy’s DSL program.

Dispatches to the DSL_FULL pipeline, or the DSL_EXTEND pipeline (pre-rules → parent strategy → post-rules) when a parent strategy is configured. Every call is wrapped in a single observability envelope that records timing, status, node count and trace for metrics and sampled persistence, regardless of success or failure.

Parameters:

  • externalGameId (Optional[str]) – External identifier of the game.

  • externalTaskId (Optional[str]) – External identifier of the task.

  • externalUserId (Optional[str]) – External identifier of the user.

  • data (Optional[dict]) – Event payload available to the program.

Returns:

tuple – (points, case_name) or (points, case_name, callback_data) produced by the program; (0, None) when the definition has no AST.

Return type:

Tuple