app.services.strategy_service module

app.services.strategy_service.is_custom_strategy_id(strategy_id)[source]

Return True when strategy_id addresses a DB-stored strategy.

Parameters:

strategy_id (str | None)

Return type:

bool

app.services.strategy_service.parse_custom_strategy_id(strategy_id)[source]

Strip the custom: prefix and return the underlying uuid.

Parameters:

strategy_id (str)

Return type:

str

app.services.strategy_service.resolve_realm_id(*, api_key=None, oauth_user_id=None)[source]

Tenant-boundary resolver shared by call sites that don’t have an AuthContext handy (e.g. UserPointsService).

Same convention as _resolve_realm_id in app/api/v1/endpoints/strategies_custom.py:

  • API key present → its value is the realm.

  • OAuth admin → falls back to configs.KEYCLOAK_REALM.

  • Neither → None (legacy unauthenticated path; any attempt to load a custom: strategy will then 404, which is the desired tenant-isolation behaviour).

Parameters:

  • api_key (str | None)

  • oauth_user_id (str | None)

Return type:

str | None

class app.services.strategy_service.StrategyService(strategy_definition_service=None, *, dsl_interpreter=None, analytics_service=None, execution_observer=None)[source]

Bases: BaseService

Service class for managing strategies.

Resolution is a two-step routing:

  • built-in strategies (registry-discovered BaseStrategy subclasses) keep their bare id, e.g. "default".

  • persistent strategies authored from the dashboard live in the strategydefinition table and are addressed as "custom:<uuid>". The resolver returns a thin descriptor for them; the DSL interpreter that runs the AST plugs into this same method.

Parameters:

  • strategy_definition_service (StrategyDefinitionService | None)

  • dsl_interpreter (DslInterpreter | None)

  • analytics_service (Any | None)

  • execution_observer (Any | None)

__init__(strategy_definition_service=None, *, dsl_interpreter=None, analytics_service=None, execution_observer=None)[source]

Initializes the StrategyService.

strategy_definition_service is optional so legacy call sites that only need built-ins still work after the custom-strategy wiring. When it’s omitted, attempting to resolve a custom: id raises NotFoundError with a clear message rather than silently crashing.

dsl_interpreter and analytics_service are required to instantiate DslStrategy for custom: ids. They are optional kwargs to preserve the legacy StrategyService() no-arg call style still in use by tests and by UserPointsService.__init__ until the container injection lands; when missing, get_strategy_instance raises a precise InternalServerError instead of crashing with AttributeError.

Parameters:

  • strategy_definition_service (StrategyDefinitionService | None)

  • dsl_interpreter (DslInterpreter | None)

  • analytics_service (Any | None)

  • execution_observer (Any | None)

Return type:

None

list_all_strategies()[source]

Lists all available built-in strategies.

Custom DB-stored strategies are returned through the dedicated /v1/strategies/custom endpoints rather than mixed in here, so the legacy contract of this endpoint stays stable.

Return type:

list[dict[str, Any]]

get_strategy_by_id(id)[source]

Retrieves a built-in strategy by its ID.

Return type:

dict[str, Any]

get_Class_by_id(id)[source]

Retrieves the instance of a built-in strategy by its ID.

Only handles the registry path; custom DSL strategies need a DB round-trip and use the async resolve() instead.

Return type:

Any

async resolve(strategy_id, *, realmId=None)[source]

Single resolution entrypoint for both code paths.

Returns a small descriptor. For built-ins:

{"kind": "BUILT_IN", "id": "default", "instance": <obj>}

For custom strategies:

{"kind": "DSL_FULL" | "DSL_EXTEND",
 "id": "custom:<uuid>",
 "definition": <StrategyDefinitionRead>}

Execution (BaseStrategy.calculate_points delegating to the DSL interpreter when the descriptor is a DSL one) plugs into this same method.

Parameters:

  • strategy_id (str) – A built-in id (e.g. "default") or a "custom:<uuid>" id.

  • realmId (str, optional) – Tenant boundary used to scope custom strategy lookups.

Returns:

dict – The resolution descriptor described above.

Raises:

NotFoundError – If a custom: id is given but custom-strategy resolution is not wired, or the definition is not found.

Return type:

dict[str, Any]

async get_strategy_instance(strategy_id, *, realmId=None)[source]

Single async entrypoint that returns something with calculate_points(...) - either a built-in registry singleton or a freshly-constructed DslStrategy wrapping a DB-persisted AST.

For non-custom: ids this delegates to the sync get_Class_by_id so existing test patches on that method keep intercepting. For custom:<uuid> it fetches the definition scoped by realmId (multi-tenant isolation is enforced at the repository layer in StrategyDefinitionService.get_strategy) and wires the same shared DslInterpreter + analytics service injected at construction.

Raises InternalServerError if the DSL collaborators were not wired - a clearer signal than AttributeError for ops.

Parameters:

  • strategy_id (str)

  • realmId (str | None)

Return type:

BaseStrategy