from fastapi import FastAPI from fastapi.responses import HTMLResponse from fastapi.requests import Request from fastapi import Query from pymongo.asynchronous.database import AsyncDatabase from extra.NodeDTO import NodeDTO, NodeShortDTO from typing import List, Annotated class WebExtension: app: FastAPI dbStore: AsyncDatabase def __init__(self, core): self.core = core self.app = core.app self.dbService = core.dbService @self.app.get(f"{self.core.context}/nodes/list") @self.core.authManager.authRequest() async def listOfNodes(request: Request, p: bool = Query(False), m:bool = Query(False), devices: List[str] = Query([])): l = await self.dbService.listOfNodes(devices) return [NodeDTO(node, p, m) for node in l] @self.app.get(f"{self.core.context}/nodes/direct") @self.core.authManager.authRequest() async def listOfDirectNodes(request: Request, p: bool = Query(False), m:bool = Query(False), devices: List[str] = Query([])): l = await self.dbService.listOfDirectNodes(devices) return [NodeDTO(node, p, m) for node in l] @self.app.get(self.core.context + "/nodes/search") async def listOfFindLikeName(name=str, devices: List[str] = Query([])): l = await self.dbService.listOfFindLikeName(name, devices) return [NodeShortDTO(node) for node in l] @self.app.get(self.core.context + "/nodes/{num}") @self.core.authManager.authRequest() async def oneNode(request: Request, num: int, p: bool = Query(False), m:bool = Query(False), devices: List[str] = Query([])): c = await self.dbService.oneNode(num, devices) if c: return NodeDTO(c, p, m) else: raise HTMLResponse(status_code=404) @self.app.get(self.core.context + "/nodes") @self.core.authManager.authRequest() async def listOfSelectedNodes(request: Request, nums: List[int] = Query(None), p: bool = Query(False), m:bool = Query(False), devices: List[str] = Query([])): if type(nums) != list: nums = [nums] l = await self.dbService.listOfSelectedNodes(nums, devices) return [NodeDTO(node, p, m) for node in l]