選択できるのは25トピックまでです。 トピックは、先頭が英数字で、英数字とダッシュ('-')を使用した35文字以内のものにしてください。

83 行
2.8 KiB

このファイルには曖昧(ambiguous)なUnicode文字が含まれています!

このファイルには曖昧(ambiguous)なUnicode文字が含まれており、あなたが使用しているロケールにおいて他の文字と混同する可能性があります。 あなたのユースケースが意図的かつ正当な場合はこの警告を無視して構いません。 それらの文字をハイライトするにはエスケープボタンを使用します。

from fastapi import FastAPI, Depends
from starlette.requests import Request
from .server_data import ServerData
from .database.db import database
# from .utils.dependencies import right_checkers, get_current_user
from .utils.dependencies import get_current_user
from .utils.responses import (
ResponseStructure,
get_base_url,
validate_response,
)
from .utils.users import check_user_rights
from .schemas.responses import GetRootResponse
from .schemas.users import User
from .routers.users import router as users_router
from .routers.commands import router as commands_router
# TODO
# 1. Добавить список отклоняемых токенов и их обработку.
# 2. Добавить предварительный вариант генерации access_tokens.
# 3. Продумать, наконец, концепцию cid и wid, в ключе их применения при
# настройке команд и запуске воркеров.
# 4. Решить сколько команд может запускать пользователь единовременно.
# 5. Добавить способ указания принадлежности воркеров и команд пользователям.
# 6. Разобраться с местом раcположения базы данных.
data = ServerData()
app = FastAPI()
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
@app.get("/", response_model=GetRootResponse, tags=["Home"])
async def get_primary_commands(request: Request,
user: User = Depends(get_current_user)):
check_user_rights(user, "read")
resp = ResponseStructure(get_base_url(request))
resp.add_data(name="Calculate Server", version="0.1")
resp.add_link("commands", "/commands?gui={is_gui}", templated=True)
resp.add_link("find_command", "/commands/{console_command}?gui={is_gui}",
templated=True)
resp.add_link("workers", "/workers")
return validate_response(resp.get_dict(), GetRootResponse,
media_type="application/hal+json")
# @app.get("/workers/{wid}", tags=["Workers management"],
# dependencies=[Depends(right_checkers["write"])])
# async def get_worker(wid: int):
# '''Тестовый обработчик.'''
# worker = data.get_worker_object(wid=wid)
# worker.run()
# print(f"worker: {type(worker)} object {worker}")
# await worker.send({"text": "INFO"})
# worker_data = await worker.get()
# if worker_data['type'] == 'log':
# data.log_message[data['level']](data['msg'])
# return worker_data
# Authentification and users management.
app.include_router(users_router)
# Commands running and management.
app.include_router(commands_router)