def _load_pre_worker_entry_points()

in src/vw-serving/src/vw_serving/serve.py [0:0]

• 25 lines of code
• 8 McCabe index (conditional complexity)

    def _load_pre_worker_entry_points(cls):
        try:
            server_config_class = cls._load_class_entry_point("algorithm.serve.server_config", "config_api")
            if server_config_class:
                cls._server_config = server_config_class()
        except Exception:
            cls.app.logger.exception("Unable to load server_config entry point")

        if cls._server_config is None:
            cls._server_config = BaseServerConfig()

        cls.app.logger.info("loading entry points")
        for entry_point in iep(group="algorithm.io.data_handlers.serve"):
            cls.request_iterators[entry_point.name] = entry_point.load()
            cls.app.logger.info("loaded request iterator %s", entry_point.name)

        for entry_point in iep(group="algorithm.request_iterators"):
            warnings.warn("entrypoint algorithm.request_iterators is deprecated "
                          "in favor of algorithm.io.data_handlers.serve", DeprecationWarning)
            cls.request_iterators[entry_point.name] = entry_point.load()
            cls.app.logger.info("loaded request iterator %s", entry_point.name)

        for entry_point in iep(group="algorithm.response_encoders"):
            cls.response_encoders[entry_point.name] = entry_point.load()
            cls.app.logger.info("loaded response encoder %s", entry_point.name)

        try:
            cls._model_class = cls._load_class_entry_point("algorithm", "model")
        except Exception as e:
            raise_with_traceback(InferenceAlgorithmError("Unable to load algorithm.model entry point", caused_by=e))