Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use handler_dispatcher method in localstack

Best Python code snippet using localstack_python

app.py

Source:app.py

...147        handler_dispatcher = {148            "start_capturing": lambda: print(user_event, payload, channel_id),149            "cancel_process": lambda: print(user_event, payload, channel_id)150        }151        dispatcher = handler_dispatcher(user_event.get(152            'action', lambda: print(user_event, payload, channel_id)))153        dispatcher()154    async def default_handler(self, event):155        print(event)156    async def ari_event_handler(self, event):157        """ __ALL__ Event Handler """158        handler_dispatcher = {159            "StasisStart": self.init_handler,160            "ChannelDtmfReceived": self.dtmf_handler,161            "ChannelUserevent": self.user_event_handler,162            "ChannelDestroyed": self.channel_hangup_handler,163            "ChannelHangupRequest": self.channel_hangup_handler,164            "StasisEnd": self.end_handler,165        }...

test_dispatcher.py

Source:test_dispatcher.py

...43        router.add("/health", TestResource())44        assert router.dispatch(Request("GET", "/health")).json == {"message": "GET/OK"}45        assert router.dispatch(Request("POST", "/health")).get_data(True) == "POST/OK"46class TestHandlerDispatcher:47    def test_handler_dispatcher(self):48        router = Router(dispatcher=handler_dispatcher())49        def handler_foo(_request: Request) -> Response:50            return Response("ok")51        def handler_bar(_request: Request, bar, baz) -> Dict[str, any]:52            response = Response()53            response.set_json({"bar": bar, "baz": baz})54            return response55        router.add("/foo", handler_foo)56        router.add("/bar/<int:bar>/<baz>", handler_bar)57        assert router.dispatch(Request("GET", "/foo")).data == b"ok"58        assert router.dispatch(Request("GET", "/bar/420/ed")).json == {"bar": 420, "baz": "ed"}59        with pytest.raises(NotFound):60            assert router.dispatch(Request("GET", "/bar/asfg/ed"))61    def test_handler_dispatcher_invalid_signature(self):62        router = Router(dispatcher=handler_dispatcher())63        def handler(_request: Request, arg1) -> Response:  # invalid signature64            return Response("ok")65        router.add("/foo/<arg1>/<arg2>", handler)66        with pytest.raises(TypeError):67            router.dispatch(Request("GET", "/foo/a/b"))68    def test_handler_dispatcher_with_dict_return(self):69        router = Router(dispatcher=handler_dispatcher())70        def handler(_request: Request, arg1) -> Dict[str, Any]:71            return {"arg1": arg1, "hello": "there"}72        router.add("/foo/<arg1>", handler)73        assert router.dispatch(Request("GET", "/foo/a")).json == {"arg1": "a", "hello": "there"}74    def test_handler_dispatcher_with_text_return(self):75        router = Router(dispatcher=handler_dispatcher())76        def handler(_request: Request, arg1) -> str:77            return f"hello: {arg1}"78        router.add("/<arg1>", handler)79        assert router.dispatch(Request("GET", "/world")).data == b"hello: world"80    def test_handler_dispatcher_with_none_return(self):81        router = Router(dispatcher=handler_dispatcher())82        def handler(_request: Request):83            return None84        router.add("/", handler)...

PlayersConnector.py

Source:PlayersConnector.py

1import asyncio2from pyrogram import Client, compose3class PlayersConnector:4    def __init__(self, logger, players, handler_dispatcher):5        self.logger = logger6        self.apps = []7        for player in players:8            app = Client(player.name, player.api_id, player.api_hash)9            app.add_handler(handler_dispatcher)10            self.apps.append(app)11            self.logger.log("Ð¡Ð¾ÐµÐ´Ð¸Ð½ÐµÐ½Ð¸Ðµ Ð´Ð»Ñ " + app.name + " ÑÑÐ¿ÐµÑÐ½Ð¾ ÑÐ¾Ð·Ð´Ð°Ð½Ð¾")12        self.logger.log("Ð¡Ð¾ÐµÐ´Ð¸Ð½ÐµÐ½Ð¸Ñ ÑÐ¾Ð·Ð´Ð°Ð½Ñ")13    def get_apps(self):14        return self.apps15    def start(self):16        asyncio.run(compose(self.apps))...

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.