68 lines
3.1 KiB
Python
68 lines
3.1 KiB
Python
from sqlalchemy.orm import Session
|
||
from typing import List, Optional, Dict, Any
|
||
|
||
from app.models.calculation import Calculation
|
||
from app.schemas.calculation import CalculationCreate, CalculationUpdate
|
||
from app.fuzzy_solver import solve_cutting_problem # Импортируем асинхронную заглушку
|
||
|
||
# Делаем функцию создания асинхронной
|
||
async def create_calculation(db: Session, calc_in: CalculationCreate) -> Calculation:
|
||
"""
|
||
Асинхронно создает запись о расчете, вызывает заглушку и сохраняет результат.
|
||
"""
|
||
# 1. Вызываем асинхронную заглушку расчета
|
||
print("Calling async solver...")
|
||
output_results = await solve_cutting_problem(calc_in.input_params)
|
||
print("Async solver finished.")
|
||
|
||
# 2. Определяем другие параметры (логика остается)
|
||
objective_score = output_results.get("waste_percentage")
|
||
model_name = calc_in.model_name or "default_fuzzy_model_v1"
|
||
|
||
# 3. Создаем объект модели SQLAlchemy
|
||
# Операции с БД остаются синхронными в этом примере
|
||
# Для полной асинхронности потребовался бы async-драйвер и AsyncSession
|
||
print("Creating DB object...")
|
||
db_calc = Calculation(
|
||
input_params=calc_in.input_params,
|
||
output_results=output_results,
|
||
objective_score=objective_score,
|
||
model_name=model_name
|
||
)
|
||
|
||
# 4. Сохраняем в базу данных (синхронные операции)
|
||
print("Adding to DB session...")
|
||
db.add(db_calc)
|
||
print("Committing DB session...")
|
||
db.commit()
|
||
print("Refreshing DB object...")
|
||
db.refresh(db_calc)
|
||
print("Calculation created and saved.")
|
||
return db_calc
|
||
|
||
# Функции чтения можно оставить синхронными, т.к. они быстрые
|
||
# и используют синхронную сессию
|
||
def get_calculation_by_id(db: Session, calc_id: int) -> Optional[Calculation]:
|
||
"""
|
||
Получает расчет по его ID.
|
||
"""
|
||
return db.query(Calculation).filter(Calculation.id == calc_id).first()
|
||
|
||
def get_calculations(db: Session, skip: int = 0, limit: int = 100) -> List[Calculation]:
|
||
"""
|
||
Получает список всех расчетов с пагинацией.
|
||
"""
|
||
return db.query(Calculation).offset(skip).limit(limit).all()
|
||
|
||
# Опционально: Функция для обновления расчета (если нужно)
|
||
# def update_calculation(db: Session, calc_id: int, calc_update: CalculationUpdate) -> Optional[Calculation]:
|
||
# db_calc = get_calculation_by_id(db, calc_id)
|
||
# if not db_calc:
|
||
# return None
|
||
# update_data = calc_update.model_dump(exclude_unset=True) # Pydantic v2
|
||
# # update_data = calc_update.dict(exclude_unset=True) # Pydantic v1
|
||
# for key, value in update_data.items():
|
||
# setattr(db_calc, key, value)
|
||
# db.commit()
|
||
# db.refresh(db_calc)
|
||
# return db_calc |