Working with the Engee external API (Personal Access Token)
In the example, we consider working with the external Engee API using PAT (Personal Access Token) - the EngeeManager control class is implemented, the functions of checking the status, starting, stopping, executing commands, uploading and downloading files.
This example does not consider asynchronous code execution, as well as the implementation of access to the Engee API using OAuth 2.0.
Introduction
Working with Engee through an external API opens up even broader possibilities for automating technical calculations, developing and optimizing models, testing complex technical systems using computer modeling, and, most importantly, integrating with external engineering environments.
To get started, you need to read documentation to the Engee API. It also tells you how to get a Personal Access Token (PAT).
Environment preparation
When developing the work example, Python version 3.13 was used.
Install the necessary dependencies:
%pip install -r requirements.txt
After receiving the token for this project, it is enough to simply specify it in the file. .env attached to the project. When publishing a project to a public repository, don't forget to delete this file or add it. .gitignore, and also specify your PAT in the repository secrets.
The PAT specified in the project's .env has been edited to protect personal data.
EngeeManager implementation
Below is the implementation of all the necessary classes and functions. Additionally, they are provided with explanations, and in difficult-to-understand places with comments.
import os
from abc import ABC, abstractmethod
from typing import Optional, Dict, Any, Union
from dataclasses import dataclass
import requests
import json
from enum import Enum
from dotenv import load_dotenv
class ServerStatus(str, Enum):
RUNNING = "running"
STOPPED = "stopped"
STARTING = "starting"
@dataclass
class EngeeConfig:
"""Конфигурация подключения к Engee."""
pat: str
base_url: str = "https://engee.com"
timeout: float = 36000.0 # Default idle timeout, seconds
# serverName: str = "demo00000000-engeeid"
class IEngeeAPI(ABC):
"""Абстрактный интерфейс для API Engee."""
@abstractmethod
def get_server_info(self) -> Optional[Dict[str, Any]]:
pass
@abstractmethod
def start_server(self) -> Optional[Dict[str, Any]]:
pass
@abstractmethod
def stop_server(self) -> bool:
pass
@abstractmethod
def execute_command(self, command: str) -> Optional[str]:
pass
@abstractmethod
def upload_file(self, destination_path: str, local_path: str) -> bool:
pass
class EngeeAPI(IEngeeAPI):
"""Реализация API для взаимодействия с Engee."""
def __init__(self, config: EngeeConfig):
self.config = config
self._headers = {"Authorization": f"Bearer {config.pat}"}
def _make_request(self, method: str, url: str, **kwargs) -> Optional[requests.Response]:
"""Внутренний метод для выполнения HTTP-запросов."""
try:
response = requests.request(method, url, headers=self._headers, **kwargs)
return response if response.status_code < 400 else None
except requests.exceptions.RequestException as e:
print(f"Ошибка запроса: {e}")
return None
def get_server_info(self) -> Optional[Dict[str, Any]]:
"""Получить информацию о состоянии сервера."""
info_url = f"{self.config.base_url}/account/api/engee/info"
response = self._make_request("GET", info_url)
return response.json() if response else None
def start_server(self) -> Optional[Dict[str, Any]]:
"""Запустить сервер."""
info = self.get_server_info()
if not info or info.get("serverStatus") != ServerStatus.STOPPED:
print("The server is not in the 'stopped' state, startup is not possible.")
return None
start_url = f"{self.config.base_url}/account/api/engee/start"
payload = {
"url": info["url"],
"inactivityTimeout": self.config.timeout
}
headers = {**self._headers, "Content-Type": "application/json"}
response = self._make_request("POST", start_url, json=payload)
return response.json() if response else None
def stop_server(self) -> bool:
"""Остановить сервер."""
stop_url = f"{self.config.base_url}/account/api/engee/stop"
response = self._make_request("DELETE", stop_url)
success = response is not None and response.status_code == 204
if success:
print("The server is stopped")
return success
def execute_command(self, command: str) -> Optional[str]:
"""Выполнить команду на сервере."""
info = self.get_server_info()
if not info:
print("Couldn't get information about the server.")
return None
if info.get("serverStatus") != ServerStatus.RUNNING:
status = info.get("serverStatus")
if status == ServerStatus.STOPPED:
print("The server is not running. Use the `start_server()` command to start the server.")
else:
print("The server is starting up. Please try again later.")
return None
server_url = f"{self.config.base_url}{info['url']}"
eval_url = f"{server_url}/external/command/eval"
payload = {"command": command}
headers = {**self._headers, "Content-Type": "application/json"}
response = self._make_request("POST", eval_url, json=payload)
if response and response.status_code == 201:
result = response.json().get("result")
print(result)
return result
else:
print("No result")
return None
def upload_file(self, destination_path: str, local_path: str) -> bool:
"""Загрузить файл на сервер."""
if not os.path.isfile(local_path):
raise FileNotFoundError(f"Файл не найден: {local_path}")
info = self.get_server_info()
if not info:
print("Couldn't get information about the server.")
return False
if info.get("serverStatus") != ServerStatus.RUNNING:
status = info.get("serverStatus")
if status == ServerStatus.STOPPED:
print("The server is not running. Use the `start_server()` command to start the server.")
else:
print("The server is starting up. Please try again later.")
return False
server_url = f"{self.config.base_url}{info['url']}"
upload_url = f"{server_url}/external/file/upload"
filename = os.path.basename(local_path)
files = {
'paths': (None, json.dumps({"sources": [destination_path]})),
'upload_files': (filename, open(local_path, 'rb'))
}
try:
headers = {**self._headers}
response = self._make_request("POST", upload_url, files=files)
if response:
if response.status_code == 201:
print("Uploaded successfully")
return True
elif response.status_code == 200:
print(f"Файл с таким именем существует. Загруженному файлу присвоено имя {response.text}")
return True
else:
print(f"Ошибка: {response.status_code} – {response.text}")
return False
finally:
if 'upload_files' in files and hasattr(files['upload_files'][1], 'close'):
files['upload_files'][1].close()
def download(self, remote_path: Union[str, list[str]], local_save_path: str) -> bool:
"""
Универсальный метод скачивания.
:param remote_path: строка (один файл) или список строк (несколько файлов)
:param local_save_path: локальный путь для сохранения (файл или ZIP)
:return: True при успехе, False при ошибке
"""
# Determining whether one path has been passed or several
if isinstance(remote_path, str):
paths_to_download = [remote_path]
single_file_mode = True
elif isinstance(remote_path, list):
paths_to_download = remote_path
single_file_mode = False
else:
print("Error: remote_path must be a string or a list of strings.")
return False
# We check that all paths start with '/user/'
for path in paths_to_download:
if not path.startswith('/user/'):
print(f"Ошибка: Все пути должны начинаться с '/user/'. Указан: {path}")
return False
info = self.get_server_info()
if not info:
print("Couldn't get information about the server.")
return False
if info.get("serverStatus") != ServerStatus.RUNNING:
status = info.get("serverStatus")
if status == ServerStatus.STOPPED:
print("The server is not running. Use the `start_server()` command to start the server.")
else:
print("The server is starting up. Please try again later.")
return False
server_url = f"{self.config.base_url}{info['url']}"
download_url = f"{server_url}/external/file/download"
payload = {"sources": paths_to_download}
try:
response = self._make_request("POST", download_url, json=payload)
if not response:
print("The download request failed.")
return False
if response.status_code == 404:
print(f"Один или несколько файлов из списка не существуют на сервере: {paths_to_download}")
return False
if response.status_code == 403:
print(f"Один или несколько файлов нарушают правило: должны быть внутри /user/. Проверьте: {paths_to_download}")
return False
if response.status_code != 200:
print(f"Ошибка при скачивании: {response.status_code} – {response.text}")
return False
# Creating a directory to save
os.makedirs(os.path.dirname(local_save_path), exist_ok=True)
content_type = response.headers.get('Content-Type', '')
if single_file_mode:
# Download as a single file
with open(local_save_path, 'wb') as f:
f.write(response.content)
print(f"Файл '{remote_path}' успешно скачан в '{local_save_path}'")
else:
# Download as a ZIP archive
if 'application/x-zip-compressed' not in content_type and 'zip' not in content_type:
print(f"Предупреждение: Ответ не является ZIP-архивом. Content-Type: {content_type}")
with open(local_save_path, 'wb') as f:
f.write(response.content)
print(f"Архив с файлами успешно скачан в '{local_save_path}'")
return True
except Exception as e:
print(f"Ошибка при скачивании: {e}")
return False
class EngeeManager:
"""Основной класс для управления Engee"""
def __init__(self, config: EngeeConfig):
self.api = EngeeAPI(config)
def get_status(self) -> Optional[ServerStatus]:
"""Получить текущий статус сервера."""
info = self.api.get_server_info()
if info:
status_str = info.get("serverStatus")
try:
return ServerStatus(status_str)
except ValueError:
return None
return None
def start(self) -> Optional[Dict[str, Any]]:
"""Запустить сервер."""
return self.api.start_server()
def stop(self) -> bool:
"""Остановить сервер."""
return self.api.stop_server()
def command(self, command: str) -> Optional[str]:
"""Выполнить команду на сервере."""
return self.api.execute_command(command)
def upload(self, destination_path: str, local_path: str) -> bool:
"""Загрузить файл на сервер."""
return self.api.upload_file(destination_path, local_path)
def download(self, remote_path: Union[str, list[str]], local_save_path: str) -> bool:
"""
Универсальный метод скачивания: один файл или несколько.
:param remote_path: строка или список строк
:param local_save_path: куда сохранить
"""
return self.api.download(remote_path, local_save_path)
def load_config_from_env(env_file: str = ".env") -> EngeeConfig:
"""Загрузить конфигурацию из файла .env."""
try:
load_dotenv()
except ImportError:
print("The python-dotenv library is not installed. Skipping the download of .env.")
pat = os.environ.get("ENGEE_PAT")
if not pat:
raise ValueError("Переменная окружения ENGEE_PAT не найдена.")
return EngeeConfig(pat=pat)
Usage example
Download the configuration from the file .env and create a sample class EngeeManager:
config = load_config_from_env()
manager = EngeeManager(config)
Engee is stopped, let's check the status:
print(f"Текущий статус: {manager.get_status()}")
Engee Launch
Let's launch Engee and check the status:
manager.start()
print(f"Текущий статус: {manager.get_status()}")
While the session is not running, we will not be able to execute commands.:
manager.command("pwd()")
Trying to run it again will also not give results.:
manager.start()
Executing commands
Immediately after starting the Engee session, we can send commands via the API to execute:
print(f"Текущий статус: {manager.get_status()}")
pwd = manager.command("pwd()")
engee_version = manager.command("engee.version()")
Uploading files
Upload the file to the server:
upload_status = manager.upload("/user/upload_test/", r'D:\engee_api\upload_test\upload_test.txt')
The function returns a boolean value of True upon successful loading. Let's check if the uploaded file is on the server.:
files = manager.command('readdir("/user/upload_test/")')
We will try to re-upload the file with the same name.:
upload_status = manager.upload("/user/upload_test/", r'D:\engee_api\upload_test\upload_test.txt')
Contents of the target directory on the server:
files = manager.command('readdir("/user/upload_test/")')
Downloading files
Download one of the previously uploaded files to a folder on your computer:
manager.download("/user/upload_test/upload_test_1.txt", r"D:\engee_api\download_test\download_test.txt")
Let's try to download several files at once. We recall that earlier in the variable files we have recorded a list of files in the resulting directory on the server. Let's pass this list to the appropriate function.:
manager.download(
files,
r"D:\engee_api\download_test\download_test.zip"
)
We received the expected error - according to the documentation, we must specify the paths of the downloaded files, child /user/.
manager.download(
["/user/upload_test/upload_test.txt", "/user/upload_test/upload_test_1.txt"],
r"D:\engee_api\download_test\download_test.zip"
)
The files are downloaded to the computer, you can check their presence in the folder of this example.
Engee Stop
stop_status = manager.stop()
print(f"Текущий статус: {manager.get_status()}")
pwd = manager.command("pwd()")
Conclusion
We have reviewed the implementation of a class for working with the external Engee API via PAT, as well as the use cases and results of functions for checking the status, starting, stopping Engee, sending commands, and uploading/downloading files.
However, these are not all the features of the Engee API available even for the current release - for example, we did not consider [Asynchronous execution] in this example. кода](https://engee.com/helpcenter/stable/en/external-software/external-software-interface-for-engee.html#асинхронное-исполнение-кода). But this is a topic that deserves a separate and detailed consideration.