# app/services/form_service.py from sqlalchemy.orm import Session from typing import Optional, List, Dict, Any from app.models.form import Form, Field, FormField from app.schemas.form import FormCreate class FormService: def __init__(self, db: Session): self.db = db def create_form(self, form_data: FormCreate, user_id=None): """Создать новую форму с полями""" # 1. Создаем форму form = Form( name=form_data.name, description=form_data.description, version=1, is_active=True, is_published=False, created_by=user_id ) self.db.add(form) self.db.flush() # 2. Создаем или получаем поля и связываем с формой for field_data in form_data.fields: existing_field = self.db.query(Field).filter( Field.name == field_data["name"] ).first() if existing_field: field = existing_field else: field = Field( name=field_data["name"], label=field_data["label"], field_type=field_data["field_type"], placeholder=field_data.get("placeholder"), help_text=field_data.get("help_text"), field_options=field_data.get("options", {}), validation_rules=field_data.get("validation_rules", {}), field_metadata=field_data.get("metadata", {}) ) self.db.add(field) self.db.flush() form_field = FormField( form_id=form.id, field_id=field.id, order=field_data.get("order", 0), is_required=field_data.get("is_required", False) ) self.db.add(form_field) self.db.commit() self.db.refresh(form) return form def get_form_by_id(self, form_id: int, include_fields: bool = True): """Получить форму с полями""" form = self.db.query(Form).filter(Form.id == form_id).first() if not form: return None return form def get_form_with_fields(self, form_id: int): """Получить форму с полями в виде словаря (для API)""" form = self.db.query(Form).filter(Form.id == form_id).first() if not form: return None # Получаем поля формы form_fields = self.db.query(FormField).filter( FormField.form_id == form_id ).order_by(FormField.order).all() fields_list = [] for ff in form_fields: field = self.db.query(Field).filter(Field.id == ff.field_id).first() if field: fields_list.append({ "id": field.id, "name": field.name, "label": field.label, "field_type": field.field_type, "order": ff.order, "is_required": ff.is_required, "placeholder": field.placeholder, "help_text": field.help_text, "options": field.field_options or {}, "validation_rules": field.validation_rules or {}, "metadata": field.field_metadata or {}, "created_at": field.created_at.isoformat() if field.created_at else None }) # Возвращаем словарь, а не объект SQLAlchemy return { "id": form.id, "name": form.name, "description": form.description, "version": form.version, "is_active": form.is_active, "is_published": form.is_published, "created_at": form.created_at.isoformat() if form.created_at else None, "updated_at": form.updated_at.isoformat() if form.updated_at else None, "created_by": form.created_by, "fields": fields_list } def get_forms_paginated(self, page=1, per_page=20, active_only=False): """Получить список форм с пагинацией""" query = self.db.query(Form) if active_only: query = query.filter(Form.is_active == True) total = query.count() items = query.offset((page - 1) * per_page).limit(per_page).all() # Преобразуем объекты в словари items_list = [] for form in items: items_list.append({ "id": form.id, "name": form.name, "description": form.description, "version": form.version, "is_active": form.is_active, "is_published": form.is_published, "created_at": form.created_at.isoformat() if form.created_at else None, "updated_at": form.updated_at.isoformat() if form.updated_at else None, "created_by": form.created_by }) return { "items": items_list, "total": total, "page": page, "per_page": per_page, "pages": (total + per_page - 1) // per_page }