Spaces:
Build error
Build error
Validify-testbot-1
/
botbuilder-python
/tests
/functional-tests
/functionaltestbot
/flask_bot_app
/bot_app.py
# Copyright (c) Microsoft Corporation. All rights reserved. | |
# Licensed under the MIT License. | |
import asyncio | |
import sys | |
from types import MethodType | |
from flask import Flask, Response, request | |
from botbuilder.core import ( | |
BotFrameworkAdapter, | |
BotFrameworkAdapterSettings, | |
MessageFactory, | |
TurnContext, | |
) | |
from botbuilder.schema import Activity, InputHints | |
from .default_config import DefaultConfig | |
from .my_bot import MyBot | |
class BotApp: | |
"""A Flask echo bot.""" | |
def __init__(self): | |
# Create the loop and Flask app | |
self.loop = asyncio.get_event_loop() | |
self.flask = Flask(__name__, instance_relative_config=True) | |
self.flask.config.from_object(DefaultConfig) | |
# Create adapter. | |
# See https://aka.ms/about-bot-adapter to learn more about how bots work. | |
self.settings = BotFrameworkAdapterSettings( | |
self.flask.config["APP_ID"], self.flask.config["APP_PASSWORD"] | |
) | |
self.adapter = BotFrameworkAdapter(self.settings) | |
# Catch-all for errors. | |
async def on_error(adapter, context: TurnContext, error: Exception): | |
# This check writes out errors to console log .vs. app insights. | |
# NOTE: In production environment, you should consider logging this to Azure | |
# application insights. | |
print(f"\n [on_turn_error]: {error}", file=sys.stderr) | |
# Send a message to the user | |
error_message_text = "Sorry, it looks like something went wrong." | |
error_message = MessageFactory.text( | |
error_message_text, error_message_text, InputHints.expecting_input | |
) | |
await context.send_activity(error_message) | |
# pylint: disable=protected-access | |
if adapter._conversation_state: | |
# If state was defined, clear it. | |
await adapter._conversation_state.delete(context) | |
self.adapter.on_turn_error = MethodType(on_error, self.adapter) | |
# Create the main dialog | |
self.bot = MyBot() | |
def messages(self) -> Response: | |
"""Main bot message handler that listens for incoming requests.""" | |
if "application/json" in request.headers["Content-Type"]: | |
body = request.json | |
else: | |
return Response(status=415) | |
activity = Activity().deserialize(body) | |
auth_header = ( | |
request.headers["Authorization"] | |
if "Authorization" in request.headers | |
else "" | |
) | |
async def aux_func(turn_context): | |
await self.bot.on_turn(turn_context) | |
try: | |
task = self.loop.create_task( | |
self.adapter.process_activity(activity, auth_header, aux_func) | |
) | |
self.loop.run_until_complete(task) | |
return Response(status=201) | |
except Exception as exception: | |
raise exception | |
def test() -> Response: | |
""" | |
For test only - verify if the flask app works locally - e.g. with: | |
```bash | |
curl http://127.0.0.1:3978/api/test | |
``` | |
You shall get: | |
``` | |
test | |
``` | |
""" | |
return Response(status=200, response="test\n") | |
def run(self, host=None) -> None: | |
try: | |
self.flask.run( | |
host=host, debug=False, port=self.flask.config["PORT"] | |
) # nosec debug | |
except Exception as exception: | |
raise exception | |