<reponame>landmaj/qbot<filename>qbot/app.py
import asyncio
import logging

from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
from starlette.applications import Starlette
from starlette.background import BackgroundTask
from starlette.middleware import Middleware
from starlette.requests import Request
from starlette.responses import PlainTextResponse, Response
from starlette.routing import Route

from qbot.core import registry
from qbot.event import process_slack_event
from qbot.plugins.cmc import comic_loop
from qbot.utils import verify_signature
from zinc_sdk.integrations.asgi import ZincAsgiMiddleware


async def slack(request: Request) -> Response:
    try:
        if not await verify_signature(request):
            return PlainTextResponse("Incorrect signature.", 403)
    except Exception:
        logging.exception("Could not verify signature.")
        return PlainTextResponse("Could not verify signature.", 400)

    data = await request.json()
    if "challenge" in data:
        return PlainTextResponse(data["challenge"])

    task = BackgroundTask(
        process_slack_event, event=data["event"], event_id=data["event_id"]
    )
    return PlainTextResponse("OK", background=task)


async def ping(request: Request) -> Response:
    await request.body()
    try:
        await registry.database.execute("SELECT 1")
    except Exception:
        return PlainTextResponse("Oops...", status_code=500)
    return PlainTextResponse(f"Pong!")


app = Starlette(
    routes=[Route("/", slack, methods=["POST"]), Route("/ping", ping, methods=["GET"])],
    middleware=[
        Middleware(SentryAsgiMiddleware),
        Middleware(ZincAsgiMiddleware, excluded_paths={"/ping"}),
    ],
)


@app.on_event("startup")
async def setup_registry():
    await registry.setup()


@app.on_event("startup")
async def start_comic_loop():
    if not registry.testing:
        asyncio.create_task(comic_loop())


@app.on_event("shutdown")
async def teardown_registry():
    await registry.teardown()

import asyncio
import logging

from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
from starlette.applications import Starlette
from starlette.background import BackgroundTask
from starlette.middleware import Middleware
from starlette.requests import Request
from starlette.responses import PlainTextResponse, Response
from starlette.routing import Route

from qbot.core import registry
from qbot.event import process_slack_event
from qbot.plugins.cmc import comic_loop
from qbot.utils import verify_signature
from zinc_sdk.integrations.asgi import ZincAsgiMiddleware


async def slack(request: Request) -> Response:
    try:
        if not await verify_signature(request):
            return PlainTextResponse("Incorrect signature.", 403)
    except Exception:
        logging.exception("Could not verify signature.")
        return PlainTextResponse("Could not verify signature.", 400)

    data = await request.json()
    if "challenge" in data:
        return PlainTextResponse(data["challenge"])

    task = BackgroundTask(
        process_slack_event, event=data["event"], event_id=data["event_id"]
    )
    return PlainTextResponse("OK", background=task)


async def ping(request: Request) -> Response:
    await request.body()
    try:
        await registry.database.execute("SELECT 1")
    except Exception:
        return PlainTextResponse("Oops...", status_code=500)
    return PlainTextResponse(f"Pong!")


app = Starlette(
    routes=[Route("/", slack, methods=["POST"]), Route("/ping", ping, methods=["GET"])],
    middleware=[
        Middleware(SentryAsgiMiddleware),
        Middleware(ZincAsgiMiddleware, excluded_paths={"/ping"}),
    ],
)


@app.on_event("startup")
async def setup_registry():
    await registry.setup()


@app.on_event("startup")
async def start_comic_loop():
    if not registry.testing:
        asyncio.create_task(comic_loop())


@app.on_event("shutdown")
async def teardown_registry():
    await registry.teardown()