mt5-manager/main.py

'''
File: main.py
Description: Веб-приложение для веб-сервера терминалов, основной файл FastAPI
'''

__version__ = '0.1.0'


# Импортируем нужные классы из библиотек
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, Path
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from pydantic_settings import BaseSettings, JsonConfigSettingsSource, SettingsConfigDict
from pydantic import ConfigDict
import os


from mt5_control import MT5_Control


class Config(BaseSettings):
    terminals: dict
    mt5_folder: str
    mt5_exe: str

    model_config = SettingsConfigDict()

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls,
        init_settings,
        env_settings,
        dotenv_settings,
        file_secret_settings,
    ):
        # Получаем путь к файлу из переменной окружения
        config_file = os.getenv("CONFIG_FILE", "config.json")
        return (
            init_settings,
            JsonConfigSettingsSource(settings_cls, json_file=config_file),
            env_settings,
            dotenv_settings,
            file_secret_settings,
        )


@asynccontextmanager
async def lifespan(app: FastAPI):
    config = Config(_env_file=None)
    app.state.terminals = config.terminals
    app.state.instances = {}

    for folder in app.state.terminals:
        app.state.terminals[folder]['name'] = app.state.terminals[folder].get('name', folder)

    global control
    control = MT5_Control(app.state.terminals, config.mt5_folder, config.mt5_exe)
    yield

# Создаём объект приложения
app = FastAPI(lifespan=lifespan)

# Подключаем статические файлы
app.mount('/static', StaticFiles(directory='static'), name='static')

# Создаём объект для работы с шаблонами HTML-кода
templates = Jinja2Templates(directory='templates')


@app.get('/', response_class=HTMLResponse)
async def index(request: Request):
    '''Обработчик GET / (корневой каталог) - панель управления терминалами'''
    instances = control.load_instances()
    # for name in instances:
    #     if instances[name]['pid']:
    #         instances[name]['info'] = control.instance_info(name)
    # print(instances)
    return templates.TemplateResponse('index.html', {'request': request, 'instances': instances})


@app.post('/instances/{name}')
async def info_instance(name: str = Path(..., description="Информация об экземпляре", example="MetaTrader5.1")):
    '''Полусение информации об экземпляре терминала'''
    result = control.instance_info(name)
    return JSONResponse(result)


@app.post('/start/{name}')
async def start_instance(name: str = Path(..., description="Название экземпляра", example="MetaTrader5.1")):
    '''Запуск экземпляра терминала'''
    result = control.start_mt5(name)
    return JSONResponse(result)


@app.post('/stop/{name}')
async def stop_instance(name: str = Path(..., description="Название экземпляра", example="MetaTrader5.1")):
    '''Остановка экземпляра терминала'''
    result = control.stop_mt5(name)
    return JSONResponse(result)


@app.post('/create/{name}')
async def create_instance(name: str = Path(..., description="Название экземпляра", example="MetaTrader5.1")):
    '''Создание экземпляра терминала'''
    result = control.create_mt5(name)
    return JSONResponse(result)