You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
65 lines
2.8 KiB
65 lines
2.8 KiB
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]
|