Spaces:
Runtime error
Runtime error
File size: 2,394 Bytes
8ec2c5a 307cacc 9fd6e20 ce323cf 307cacc 9fd6e20 8ec2c5a 9fd6e20 307cacc 9fd6e20 64566ca c6a2a56 307cacc 8ec2c5a ce323cf c6a2a56 ce323cf 307cacc ce323cf 9fd6e20 8ec2c5a f40718d 3300601 f40718d 3300601 8ec2c5a 9fd6e20 8ec2c5a 9fd6e20 8ec2c5a ce323cf 307cacc ce323cf 307cacc ce323cf 8ec2c5a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from asyncio import run as run_async
from contextlib import asynccontextmanager
from dependency_injector.wiring import inject, Provide
from fastapi import FastAPI, HTTPException, Depends
from loguru import logger
from typing import Any, AsyncGenerator
from slack_bolt.async_app import AsyncApp
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
from starlette.requests import Request
from starlette.responses import Response
from threading import Thread
from typing import Any, Dict, Self
from ctp_slack_bot.containers import Container
from ctp_slack_bot.core.config import Settings
from ctp_slack_bot.core.logging import setup_logging
from ctp_slack_bot.core.response_rendering import PrettyJSONResponse
from ctp_slack_bot.tasks import start_scheduler, stop_scheduler
async def main() -> None:
container = Container()
container.wire(packages=['ctp_slack_bot'])
# Setup logging.
setup_logging(container)
logger.info("Starting application")
# Start the scheduler.
scheduler = start_scheduler(container)
logger.info("Started scheduler")
# Initialize primordial dependencies in container.
container.primordial_services()
# Start Slack socket mode in a background thread and set up an event handler for the Bolt app.
bolt_app = container.slack_bolt_app()
slack_service = container.slack_service()
@bolt_app.event("message")
async def handle_message(body: Dict[str, Any]) -> None:
logger.debug("Ignored regular message: {}", body.get("event").get("text"))
#await slack_service.process_message(body)
@bolt_app.event("app_mention")
async def handle_app_mention(body: Dict[str, Any]) -> None:
#logger.debug("Ignored app mention: {}", body.get("event").get("text"))
await slack_service.process_message(body)
# Start Socket Mode handler in a background thread
socket_mode_handler = AsyncSocketModeHandler(
app=bolt_app,
app_token=container.settings().SLACK_APP_TOKEN.get_secret_value()
)
logger.info("Starting Slack Socket Mode handler…")
await socket_mode_handler.start_async()
# Shutdown.
logger.info("Shutting down application")
stop_scheduler(scheduler)
logger.info("Stopped scheduler")
if __name__ == "__main__":
# run()
run_async(main())
|