mephisto.abstractions.providers.mturk_sandbox.sandbox_mturk_requester

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_requester import MTurkRequester
from mephisto.abstractions.providers.mturk_sandbox.provider_type import PROVIDER_TYPE

from typing import Any, Optional, Mapping, TYPE_CHECKING

if TYPE_CHECKING:
    from mephisto.abstractions.database import MephistoDB
    from mephisto.data_model.requester import Requester
    from mephisto.abstractions.providers.mturk.mturk_datastore import MTurkDatastore


class SandboxMTurkRequester(MTurkRequester):
    """Wrapper around regular requester that handles removing the appended "sandbox" name"""

    # 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
        )
        # Use _requester_name to preserve sandbox behavior which
        # utilizes a different requester_name
        assert self.requester_name.endswith(
            "_sandbox"
        ), f"{self.requester_name} is not a sandbox requester"
        self._requester_name = self.requester_name[:-8]

    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)

    @classmethod
    def is_sandbox(cls) -> bool:
        """
        Determine if this is a requester on sandbox
        """
        return True

    # Required functions for a Requester implementation

    @staticmethod
    def new(db: "MephistoDB", requester_name: str) -> "Requester":
        if not requester_name.endswith("_sandbox"):
            requester_name += "_sandbox"
        return SandboxMTurkRequester._register_requester(
            db, requester_name, PROVIDER_TYPE
        )

# class SandboxMTurkRequester(mephisto.abstractions.providers.mturk.mturk_requester.MTurkRequester):

View Source

class SandboxMTurkRequester(MTurkRequester):
    """Wrapper around regular requester that handles removing the appended "sandbox" name"""

    # 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
        )
        # Use _requester_name to preserve sandbox behavior which
        # utilizes a different requester_name
        assert self.requester_name.endswith(
            "_sandbox"
        ), f"{self.requester_name} is not a sandbox requester"
        self._requester_name = self.requester_name[:-8]

    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)

    @classmethod
    def is_sandbox(cls) -> bool:
        """
        Determine if this is a requester on sandbox
        """
        return True

    # Required functions for a Requester implementation

    @staticmethod
    def new(db: "MephistoDB", requester_name: str) -> "Requester":
        if not requester_name.endswith("_sandbox"):
            requester_name += "_sandbox"
        return SandboxMTurkRequester._register_requester(
            db, requester_name, PROVIDER_TYPE
        )

Wrapper around regular requester that handles removing the appended "sandbox" name

# SandboxMTurkRequester( 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,
    ) -> "Requester":
        """
        The new method is overridden to be able to automatically generate
        the expected Requester class without needing to specifically find it
        for a given db_id. As such it is impossible to create a base Requester
        as you will instead be returned the correct Requester class according to
        the crowdprovider associated with this Requester.
        """
        from mephisto.operations.registry import get_crowd_provider_from_type

        if cls == Requester:
            # We are trying to construct a Requester, find what type to use and
            # create that instead
            if row is None:
                row = db.get_requester(db_id)
            assert row is not None, f"Given db_id {db_id} did not exist in given db"
            correct_class = get_crowd_provider_from_type(
                row["provider_type"]
            ).RequesterClass
            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 Requester class without needing to specifically find it for a given db_id. As such it is impossible to create a base Requester as you will instead be returned the correct Requester class according to the crowdprovider associated with this Requester.

# PROVIDER_TYPE = 'mturk_sandbox'

@classmethod

def is_sandbox(cls) -> bool:

View Source

    @classmethod
    def is_sandbox(cls) -> bool:
        """
        Determine if this is a requester on sandbox
        """
        return True

Determine if this is a requester on sandbox

@staticmethod

def new( db: mephisto.abstractions.database.MephistoDB, requester_name: str ) -> mephisto.data_model.requester.Requester:

View Source

    @staticmethod
    def new(db: "MephistoDB", requester_name: str) -> "Requester":
        if not requester_name.endswith("_sandbox"):
            requester_name += "_sandbox"
        return SandboxMTurkRequester._register_requester(
            db, requester_name, PROVIDER_TYPE
        )

Try to create a new requester by this name, raise an exception if the name already exists.

Implementation should call _register_requester(db, requester_id) when sure the requester can be successfully created to have it put into the db and return the result.

Inherited Members

mephisto.abstractions.providers.mturk.mturk_requester.MTurkRequester: ArgsClass; register; is_registered; get_available_budget
mephisto.data_model.requester.Requester: get_task_runs; get_total_spend; to_dict
mephisto.data_model._db_backed_meta.MephistoDataModelComponentMixin: get; async_get