Module server.stats.achievement_service

Classes

class AchievementService (message_queue_service: MessageQueueService)

All services should inherit from this class.

Services are singleton objects which manage some server task.

Expand source code 
@with_logger
class AchievementService(Service):
    def __init__(self, message_queue_service: MessageQueueService):
        self.message_queue_service = message_queue_service

    async def execute_batch_update(self, player_id, queue):
        """
        Sends a batch of achievement updates.

        # Params
        - `player_id`: the player to update the achievements for
        - `queue`: an array of achievement updates in the form:
        ```
        [{
            "achievementId": string,
            "operation": "REVEAL|INCREMENT|UNLOCK|SET_STEPS_AT_LEAST",
            "steps": integer
        }]
        ```
        `steps` is mandatory for update type `INCREMENT` and
        `SET_STEPS_AT_LEAST`.

        # Returns
        If successful, this method returns an array with the following structure:
        ```
        [{
            "achievement_id": string,
            "current_state": string,
            "current_steps": integer,
            "newly_unlocked": boolean
        }]
        ```
        Otherwise, it returns None
        """
        self._logger.info("Updating %d achievements for player %d", len(queue), player_id)
        update_data = [
            {"playerId": player_id, **data}
            for data in queue
        ]
        await self.message_queue_service.publish_many(
            config.MQ_EXCHANGE_NAME,
            "request.achievement.update",
            update_data,
        )

    def unlock(self, achievement_id, queue):
        """
        Enqueues an achievement update that reveals an achievement.

        # Params
        - `achievement_id`: the achievement to unlock
        - `queue`: the queue to put this update into so it can be batch
        executed later
        """
        queue.append({
            "achievementId": achievement_id,
            "operation": "UNLOCK"
        })

    def reveal(self, achievement_id, queue):
        """
        Enqueues an achievement update that reveals an achievement.

        # Params
        - `achievement_id`: the achievement to unlock
        - `queue`: the queue to put this update into so it can be batch
        executed later
        """
        queue.append({
            "achievementId": achievement_id,
            "operation": "REVEAL"
        })

    def increment(self, achievement_id, steps, queue):
        """
        Enqueues an achievement update that increments an achievement.

        # Params
        - `achievement_id`: the achievement to unlock
        - `steps`: the number of steps to increment
        - `queue`: the queue to put this update into so it can be batch
        executed later
        """
        if steps == 0:
            return

        queue.append({
            "achievementId": achievement_id,
            "operation": "INCREMENT",
            "steps": steps
        })

    def set_steps_at_least(self, achievement_id, steps, queue):
        """
        Enqueues an achievement update that sets the steps to the specified
        minimum steps.

        # Params
        - `achievement_id`: the achievement to update
        - `steps`: the minimum number of steps to set
        - `queue`: the queue to put this update into so it can be batch
        executed later
        """
        if steps == 0:
            return

        queue.append({
            "achievementId": achievement_id,
            "operation": "SET_STEPS_AT_LEAST",
            "steps": steps
        })

Ancestors

Methods

async def execute_batch_update(self, player_id, queue)

Sends a batch of achievement updates.

Params

  • player_id: the player to update the achievements for
  • queue: an array of achievement updates in the form:
[{
    "achievementId": string,
    "operation": "REVEAL|INCREMENT|UNLOCK|SET_STEPS_AT_LEAST",
    "steps": integer
}]

steps is mandatory for update type INCREMENT and SET_STEPS_AT_LEAST.

Returns

If successful, this method returns an array with the following structure:

[{
    "achievement_id": string,
    "current_state": string,
    "current_steps": integer,
    "newly_unlocked": boolean
}]

Otherwise, it returns None

def increment(self, achievement_id, steps, queue)

Enqueues an achievement update that increments an achievement.

Params

  • achievement_id: the achievement to unlock
  • steps: the number of steps to increment
  • queue: the queue to put this update into so it can be batch executed later
def reveal(self, achievement_id, queue)

Enqueues an achievement update that reveals an achievement.

Params

  • achievement_id: the achievement to unlock
  • queue: the queue to put this update into so it can be batch executed later
def set_steps_at_least(self, achievement_id, steps, queue)

Enqueues an achievement update that sets the steps to the specified minimum steps.

Params

  • achievement_id: the achievement to update
  • steps: the minimum number of steps to set
  • queue: the queue to put this update into so it can be batch executed later
def unlock(self, achievement_id, queue)

Enqueues an achievement update that reveals an achievement.

Params

  • achievement_id: the achievement to unlock
  • queue: the queue to put this update into so it can be batch executed later

Inherited members