How to use require_service method in localstack

Best Python code snippet using localstack_python

views.py

Source:views.py

1""" Server views for evaluation.2"""3import logging4from davisinteractive.third_party import mask_api5from django.conf import settings6from django.http import HttpResponse7from django.views.decorators.csrf import csrf_exempt8from django.views.decorators.http import require_GET, require_POST9from .decorators import authorize, json_api, require_service10from .models import Session11logger = logging.getLogger(__name__)12@json_api13@require_GET14def get_health(_):15    """ Return Healt status.16    """17    # Check the DB connection18    try:19        from django.db import connections20        for name in connections:21            cursor = connections[name].cursor()22            cursor.execute("SELECT 1;")23            row = cursor.fetchone()24            if row is None:25                return HttpResponse("db: invalid response", status=500)26    except Exception as e:27        logger.exception(e)28        return HttpResponse('db: cannot connect to database.', status=500)29    return {30        'health': 'OK',31        'name': 'DAVIS Interactive Server',32        'magic': 23,33        'evaluation_parameters': {34            'subset': settings.EVALUATION_SUBSET,35            'max_time': settings.EVALUATION_MAX_TIME,36            'max_interactions': settings.EVALUATION_MAX_INTERACTIONS,37            'metric_to_optimize': settings.EVALUATION_METRIC_TO_OPTIMIZE38        }39    }40@json_api41@require_GET42@require_service43def get_dataset_samples(_, service):44    """ Return the dataset samples.45    """46    response = service.get_samples()47    return response48@json_api49@require_GET50@require_service51def get_scribble(_, sequence, scribble_idx, service, **kwargs):52    """ Return the scribble asked.53    """54    response = service.get_scribble(sequence, scribble_idx)55    return response56@csrf_exempt57@json_api58@require_POST59@require_service60@authorize61def post_predicted_masks(request, service, user_key, session_key):62    """ Post the predicted masks and return a new scribble.63    """64    params = request.json65    params['pred_masks'] = mask_api.decode_batch_masks(params['pred_masks'])66    params['user_key'] = user_key67    params['session_key'] = session_key68    logger.info('[Session {}] Sequence: {}/{}\tPred masks shape: {}'.format(69        session_key[:8], params['sequence'], params['scribble_idx'],70        params['pred_masks'].shape))71    logger.info('[Session {}] User key: {}\tInteraction: {}'.format(72        session_key[:8], user_key[:8], params['interaction']))73    response = service.post_predicted_masks(**params)74    return response75@json_api76@require_GET77@require_service78@authorize79def get_report(_, service, session_key, user_key=None):80    """ Return the report for a single session.81    """82    df = service.get_report(session_id=session_key).copy()83    if len(df) > 0:84        df = df.groupby([85            'session_id', 'sequence', 'scribble_idx', 'interaction', 'object_id'86        ]).mean()87        df = df.drop(columns='frame')88        df = df.reset_index()89    return df.to_dict()90@csrf_exempt91@json_api92@require_POST93@require_service94@authorize95def post_finish(_, service, session_key, user_key=None):96    """ Notify the session has finished.97    Will mask the session as completed.98    Returns the generated global summary.99    """100    session = Session.objects.get(session_id=session_key)101    report = service.get_report(session_id=session_key)102    summary = service.summarize_report(report)103    session.mark_completed(summary)104    session.save()...

wandb_require.py

Source:wandb_require.py

...21            tuple([features]) if isinstance(features, str) else tuple(features)22        )23    def require_require(self) -> None:24        pass25    def _require_service(self) -> None:26        os.environ[REQUIRE_SERVICE] = "True"27        wandb.teardown = wandb._teardown  # type: ignore28        wandb.attach = wandb._attach  # type: ignore29        wandb_run.Run.detach = wandb_run.Run._detach  # type: ignore30    def require_service(self) -> None:31        disable_service = os.environ.get(_DISABLE_SERVICE)32        if disable_service:33            if REQUIRE_SERVICE in os.environ:34                del os.environ[REQUIRE_SERVICE]35            return36        self._require_service()37    def _require_report_editing(self) -> None:38        os.environ["WANDB_REQUIRE_REPORT_EDITING_V0"] = "True"39        wandb.termwarn("This is an experimental feature -- use with caution!")40    def require_report_editing(self) -> None:41        self._require_report_editing()42    def apply(self) -> None:43        """Call require_* method for supported features."""44        last_message: str = ""45        for feature_item in self._features:46            full_feature = feature_item.split("@", 2)[0]47            feature = full_feature.split(":", 2)[0]48            func_str = "require_{}".format(feature.replace("-", "_"))49            func = getattr(self, func_str, None)50            if not func:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.