/ DOCS
Evolution
Тема интерфейса

Python Flask App для управления задачами обучения

Это приложение на Flask предоставляет базовый интерфейс для создания, мониторинга и получения логов задач обучения, используя API Cloud.ru.

Перед началом работы

Необходимо настроить окружение. Для этого создайте файл конфигурации .env_demo в корневой директории проекта с следующим содержимым:

env
MLS_APIKEY_ID=
MLS_APIKEY_SECRET=
WORKSPACE_ID=
X_API_KEY=
REGION=
OUTPUT=

Зависимости

  • Flask

  • Pydantic

  • Requests

  • urllib3

Запуск приложения

  1. Подготовьте переменные окружения.

  2. Инициализируйте TrainingJobApi.

  3. Выполните flask run.

Код приложения на flask
 import requests
 import urllib3
 from flask import Flask
 from flask import jsonify
 from flask import render_template
 from flask import request
 from flask import Response
 from pydantic.v1 import BaseSettings


 from mls_core import TrainingJobApi




 class Settings(BaseSettings):
     """Конфигурационные настройки для приложения, загружаемые из файла окружения.


     Атрибуты:
     - MLS_APIKEY_ID (str): Идентификатор API ключа для доступа к MLS (MLS - Machine Learning Space).
     - MLS_APIKEY_SECRET (str): Секретный ключ API для доступа к MLS.
     - WORKSPACE_ID (str): Идентификатор рабочего пространства в MLS.
     - X_API_KEY (str): API ключ для дополнительного доступа к другим сервисам.
     - REGION (str): Регион для запуска и выполнения задач машинного обучения.
     - OUTPUT (str): Способ вывода результатов выполнения задач.
     """
     MLS_APIKEY_ID: str
     MLS_APIKEY_SECRET: str
     WORKSPACE_ID: str
     X_API_KEY: str
     REGION: str
     OUTPUT: str


     class Config:
         """Конфигурация.


         - env_file (str): Путь к файлу окружения, откуда загружаются настройки. По умолчанию '.env_demo'.
         - env_file_encoding (str): Кодировка файла окружения. По умолчанию 'utf-8'.
         """
         env_file = '.env_demo'
         env_file_encoding = 'utf-8'




 settings = Settings()
 app = Flask(__name__)




 client = TrainingJobApi(
     'https://.../public/v2',
     settings.MLS_APIKEY_ID,
     settings.MLS_APIKEY_SECRET,
     settings.WORKSPACE_ID,
     settings.X_API_KEY,
     backoff_factor=2,
     connect_timeout=300,
     read_timeout=300,
 )




 @app.route('/', methods=['GET'])
 def home():
     """Домашняя страница приложения."""
     return render_template('example.html')




 @app.route('/create', methods=['POST'])
 def create_task():
     """Метод постановки задачи обучения в очередь на исполнение."""
     task_name = client.run_job(
         payload={
             'script': 'sleep 30; echo abc; sleep 30; echo abc; sleep 30; echo abc; sleep 30; echo abc; sleep 30; echo abc;',
             'base_image': 'cr.ai.cloud.ru/aicloud-base-images/py3.10-torch2.1.2:0.0.40',
             'instance_type': 'a100.1gpu',
             'region': settings.REGION,
             'type': 'pytorch2',
             'n_workers': 1,
             'job_desc': 'Все хорошо закончилось',
         },
     ).get('job_name')
     if task_name:
         return jsonify({'task_name': task_name})
     return {}




 @app.route('/status', methods=['POST'])
 def status():
     """Метод получения статуса задачи."""
     task_name = request.json.get('task_name')
     if task_name:
         res = client.get_job_status(name=task_name, region=settings.REGION)
         return jsonify({'status': res.get('status'), 'task_name': res.get('job_name')})
     return {}




 @app.route('/logs', methods=['POST'])
 def logs():
     """Метод получения журнала событий в потоке."""
     task_name = request.json.get('task_name')
     try:
         return Response(client.stream_logs(task_name, settings.REGION))
     except (urllib3.exceptions.ReadTimeoutError, requests.exceptions.ConnectionError):
         return {}




 if __name__ == '__main__':
     app.run(debug=True)
Код html-страницы для приложения
 <!DOCTYPE html>
 <html>
 <head>
     <title>Запуск задачи</title>
     <link href="https://maxcdn.bootstrapcdn.com/bootstrap/3.3.7/css/bootstrap.min.css" rel="stylesheet">
     <script src="https://ajax.googleapis.com/ajax/libs/jquery/3.5.1/jquery.min.js"></script>
 </head>
 <body>


 <div class="container">
     <!-- Верхняя часть: Форма -->
     <div class="row">
         <div class="col-md-12">
             <form id="myForm">
                 <div class="form-group">
                     <label>Тип инстанса:</label>
                     <select class="form-control" name="instanceType">
                         <option value="small">Small</option>
                         <option value="large">Large</option>
                         <option value="xlarge">XLarge</option>
                     </select>
                 </div>
                 <button type="submit" class="btn btn-primary">Отправить</button>
             </form>
         </div>
     </div>


     <!-- Нижняя часть: Текстовое поле для AJAX ответа -->
     <div class="row" style="margin-top: 20px;">
         <div class="col-md-12">
             <div id="result2" class="well">
                 <!-- AJAX ответ будет здесь -->
             </div>
             <div id="result" class="well">
                 <!-- AJAX ответ будет здесь -->
             </div>
         </div>
     </div>
 </div>


 <script>
     function checkTaskStatus(data) {
     $.ajax({
         url: "/status",
         type: "POST",
         data: data,
         success: function(response) {
             $('#result2').html(response.status + " " + response.task_name);
             if (["failed", "completed"].includes(response.status)) {
                 $('#result').append('<br>' + 'Задача завершилась');
             } else if (response.status === "running") {
                 // Если задача в статусе "Running", убедимся, что не начинаем загрузку логов повторно
                 // Для этого можно использовать флаг или проверять текущее состояние контента,
                 // например, запущена ли функция streamLogs для этой задачи.
                 // Пример с флагом:
                 if (!window.isStreamingLogs) {
                     window.isStreamingLogs = true; // Устанавливаем флаг
                     streamLogs(data.task_name);
                 } else {
                     window.isStreamingLogs = true;
                     checkTaskStatus(data);
                 }
             } else {
                 setTimeout(function() {checkTaskStatus(data)}, 4000);
             }
         }
     });
 }


 async function streamLogs (task_name){
     const response = await fetch(
         "/logs",
         {
         headers: {
         Accept: 'text/plain',
         'Content-Type': 'application/json',
         },
         method: 'POST',
         body: JSON.stringify({"task_name": task_name})
         }
     );


     for await (const chunk of response.body) {
         const utf8Decoder = new TextDecoder();
         $('#result').append('<br>' + utf8Decoder.decode(chunk.buffer));
     }


     // После завершения чтения логов сбрасываем флаг, чтобы можно было начать поток снова
     // при необходимости (например, если поток прервался и задача всё ещё в статусе "Running").
     window.isStreamingLogs = true;
     checkTaskStatus({"task_name": task_name})
 }


 $(document).on("submit", "#myForm", function(e) {
     e.preventDefault();
     var formData = $(this).serialize();


     // Убедимся, что сбросили флаг перед началом работы
     window.isStreamingLogs = false;


     $.ajax({
         url: "/create",
         type: "POST",
         data: formData,
         success: function(response) {
             checkTaskStatus({"task_name": response.task_name});
         }
     });
 });


     </script>
 </body>
 </html>
Предыдущая статья
Сценарии использования
Следующая статья
Пример использования CLI в makefile
Была ли эта статья полезна?
Поддержка Юридические документы
© 2025 Cloud.ru