66 lines
1.6 KiB
Python
66 lines
1.6 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
from contextlib import asynccontextmanager
|
||
|
|
|
||
|
|
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
|
||
|
|
from fastapi.middleware.cors import CORSMiddleware
|
||
|
|
|
||
|
|
from app.api.router import api_router
|
||
|
|
from app.core.config import settings
|
||
|
|
from app.db.sqlite import init_db
|
||
|
|
from app.tasks.polling import device_polling_loop, stop_task
|
||
|
|
from app.ws.manager import ws_manager
|
||
|
|
|
||
|
|
|
||
|
|
@asynccontextmanager
|
||
|
|
async def lifespan(_: FastAPI):
|
||
|
|
init_db()
|
||
|
|
poll_task = asyncio.create_task(device_polling_loop())
|
||
|
|
try:
|
||
|
|
yield
|
||
|
|
finally:
|
||
|
|
await stop_task(poll_task)
|
||
|
|
|
||
|
|
|
||
|
|
app = FastAPI(
|
||
|
|
title=settings.app_name,
|
||
|
|
version=settings.app_version,
|
||
|
|
lifespan=lifespan,
|
||
|
|
)
|
||
|
|
|
||
|
|
app.add_middleware(
|
||
|
|
CORSMiddleware,
|
||
|
|
allow_origins=["*"],
|
||
|
|
allow_credentials=True,
|
||
|
|
allow_methods=["*"],
|
||
|
|
allow_headers=["*"],
|
||
|
|
)
|
||
|
|
|
||
|
|
app.include_router(api_router, prefix=settings.api_prefix)
|
||
|
|
|
||
|
|
|
||
|
|
@app.get("/")
|
||
|
|
def healthcheck() -> dict:
|
||
|
|
return {"status": "ok", "service": settings.app_name}
|
||
|
|
|
||
|
|
|
||
|
|
@app.websocket(settings.realtime_ws_path)
|
||
|
|
async def realtime_ws(websocket: WebSocket) -> None:
|
||
|
|
await ws_manager.connect("real-time", websocket)
|
||
|
|
try:
|
||
|
|
while True:
|
||
|
|
await websocket.receive_text()
|
||
|
|
except WebSocketDisconnect:
|
||
|
|
ws_manager.disconnect("real-time", websocket)
|
||
|
|
|
||
|
|
|
||
|
|
@app.websocket(settings.alarm_ws_path)
|
||
|
|
async def alarm_ws(websocket: WebSocket) -> None:
|
||
|
|
await ws_manager.connect("alarm", websocket)
|
||
|
|
try:
|
||
|
|
while True:
|
||
|
|
await websocket.receive_text()
|
||
|
|
except WebSocketDisconnect:
|
||
|
|
ws_manager.disconnect("alarm", websocket)
|