mephisto.abstractions.providers.mturk_sandbox.sandbox_mturk_worker

View Source

#!/usr/bin/env python3

# Copyright (c) Meta Platforms and its affiliates.
# This source code is licensed under the MIT license found in the
# LICENSE file in the root directory of this source tree.

from mephisto.abstractions.providers.mturk.mturk_worker import MTurkWorker
from mephisto.abstractions.providers.mturk_sandbox.provider_type import PROVIDER_TYPE

from typing import Any, Mapping, Optional, TYPE_CHECKING

if TYPE_CHECKING:
    from mephisto.abstractions.providers.mturk.mturk_datastore import MTurkDatastore
    from mephisto.data_model.worker import Worker
    from mephisto.abstractions.database import MephistoDB


class SandboxMTurkWorker(MTurkWorker):
    """
    This class represents an individual - namely a person. It maintains components of ongoing identity for a user.
    """

    # Ensure inherited methods use this level's provider type
    PROVIDER_TYPE = PROVIDER_TYPE

    def __init__(
        self,
        db: "MephistoDB",
        db_id: str,
        row: Optional[Mapping[str, Any]] = None,
        _used_new_call: bool = False,
    ):
        super().__init__(db, db_id, row=row, _used_new_call=_used_new_call)
        self.datastore: "MTurkDatastore" = self.db.get_datastore_for_provider(
            self.PROVIDER_TYPE
        )
        # sandbox workers use a different name
        self._worker_name = self.worker_name[:-8]

    def grant_crowd_qualification(
        self, qualification_name: str, value: int = 1
    ) -> None:
        """
        Grant a qualification by the given name to this worker. Check the local
        MTurk db to find the matching MTurk qualification to grant, and pass
        that. If no qualification exists, try to create one.
        """
        return super().grant_crowd_qualification(qualification_name + "_sandbox", value)

    def revoke_crowd_qualification(self, qualification_name: str) -> None:
        """
        Revoke the qualification by the given name from this worker. Check the local
        MTurk db to find the matching MTurk qualification to revoke, pass if
        no such qualification exists.
        """
        return super().revoke_crowd_qualification(qualification_name + "_sandbox")

    def _get_client(self, requester_name: str) -> Any:
        """
        Get an mturk client for usage with mturk_utils
        """
        return self.datastore.get_sandbox_client_for_requester(requester_name)

    @staticmethod
    def new(db: "MephistoDB", worker_id: str) -> "Worker":
        return MTurkWorker._register_worker(db, worker_id + "_sandbox", PROVIDER_TYPE)

# class SandboxMTurkWorker(mephisto.abstractions.providers.mturk.mturk_worker.MTurkWorker):

View Source

class SandboxMTurkWorker(MTurkWorker):
    """
    This class represents an individual - namely a person. It maintains components of ongoing identity for a user.
    """

    # Ensure inherited methods use this level's provider type
    PROVIDER_TYPE = PROVIDER_TYPE

    def __init__(
        self,
        db: "MephistoDB",
        db_id: str,
        row: Optional[Mapping[str, Any]] = None,
        _used_new_call: bool = False,
    ):
        super().__init__(db, db_id, row=row, _used_new_call=_used_new_call)
        self.datastore: "MTurkDatastore" = self.db.get_datastore_for_provider(
            self.PROVIDER_TYPE
        )
        # sandbox workers use a different name
        self._worker_name = self.worker_name[:-8]

    def grant_crowd_qualification(
        self, qualification_name: str, value: int = 1
    ) -> None:
        """
        Grant a qualification by the given name to this worker. Check the local
        MTurk db to find the matching MTurk qualification to grant, and pass
        that. If no qualification exists, try to create one.
        """
        return super().grant_crowd_qualification(qualification_name + "_sandbox", value)

    def revoke_crowd_qualification(self, qualification_name: str) -> None:
        """
        Revoke the qualification by the given name from this worker. Check the local
        MTurk db to find the matching MTurk qualification to revoke, pass if
        no such qualification exists.
        """
        return super().revoke_crowd_qualification(qualification_name + "_sandbox")

    def _get_client(self, requester_name: str) -> Any:
        """
        Get an mturk client for usage with mturk_utils
        """
        return self.datastore.get_sandbox_client_for_requester(requester_name)

    @staticmethod
    def new(db: "MephistoDB", worker_id: str) -> "Worker":
        return MTurkWorker._register_worker(db, worker_id + "_sandbox", PROVIDER_TYPE)

This class represents an individual - namely a person. It maintains components of ongoing identity for a user.

# SandboxMTurkWorker( db: mephisto.abstractions.database.MephistoDB, db_id: str, row: Union[Mapping[str, Any], NoneType] = None, _used_new_call: bool = False )

View Source

    def __new__(
        cls,
        db: "MephistoDB",
        db_id: str,
        row: Optional[Mapping[str, Any]] = None,
        _used_new_call: bool = False,
    ) -> "Worker":
        """
        The new method is overridden to be able to automatically generate
        the expected Worker class without needing to specifically find it
        for a given db_id. As such it is impossible to create a base Worker
        as you will instead be returned the correct Worker class according to
        the crowdprovider associated with this Worker.
        """
        from mephisto.operations.registry import get_crowd_provider_from_type

        if cls == Worker:
            # We are trying to construct a Worker, find what type to use and
            # create that instead
            if row is None:
                row = db.get_worker(db_id)
            assert row is not None, f"Given db_id {db_id} did not exist in given db"
            correct_class: Type[Worker] = get_crowd_provider_from_type(
                row["provider_type"]
            ).WorkerClass
            return super().__new__(correct_class)
        else:
            # We are constructing another instance directly
            return super().__new__(cls)

The new method is overridden to be able to automatically generate the expected Worker class without needing to specifically find it for a given db_id. As such it is impossible to create a base Worker as you will instead be returned the correct Worker class according to the crowdprovider associated with this Worker.

# PROVIDER_TYPE = 'mturk_sandbox'

# def grant_crowd_qualification(self, qualification_name: str, value: int = 1) -> None:

View Source

    def grant_crowd_qualification(
        self, qualification_name: str, value: int = 1
    ) -> None:
        """
        Grant a qualification by the given name to this worker. Check the local
        MTurk db to find the matching MTurk qualification to grant, and pass
        that. If no qualification exists, try to create one.
        """
        return super().grant_crowd_qualification(qualification_name + "_sandbox", value)

Grant a qualification by the given name to this worker. Check the local MTurk db to find the matching MTurk qualification to grant, and pass that. If no qualification exists, try to create one.

# def revoke_crowd_qualification(self, qualification_name: str) -> None:

View Source

    def revoke_crowd_qualification(self, qualification_name: str) -> None:
        """
        Revoke the qualification by the given name from this worker. Check the local
        MTurk db to find the matching MTurk qualification to revoke, pass if
        no such qualification exists.
        """
        return super().revoke_crowd_qualification(qualification_name + "_sandbox")

Revoke the qualification by the given name from this worker. Check the local MTurk db to find the matching MTurk qualification to revoke, pass if no such qualification exists.

@staticmethod

def new( db: mephisto.abstractions.database.MephistoDB, worker_id: str ) -> mephisto.data_model.worker.Worker:

View Source

    @staticmethod
    def new(db: "MephistoDB", worker_id: str) -> "Worker":
        return MTurkWorker._register_worker(db, worker_id + "_sandbox", PROVIDER_TYPE)

Create a new worker attached to the given identifier, assuming it doesn't already exist in the database.

Implementation should return the result of _register_worker when sure the worker can be successfully created to have it put into the db.

Inherited Members

mephisto.abstractions.providers.mturk.mturk_worker.MTurkWorker: get_from_mturk_worker_id; get_mturk_worker_id; bonus_worker; block_worker; unblock_worker; is_blocked; is_eligible
mephisto.data_model.worker.Worker: get_agents; new_from_provider_data; get_granted_qualification; is_disqualified; is_qualified; revoke_qualification; grant_qualification; register
mephisto.data_model._db_backed_meta.MephistoDataModelComponentMixin: get; async_get