98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
from typing import Optional, List, Dict, Any
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import and_, or_
|
||
from app.repositories.base import BaseRepository
|
||
from app.models.form import Form, FormField, Field
|
||
|
||
|
||
class FormRepository(BaseRepository[Form]):
|
||
def __init__(self, db: Session):
|
||
super().__init__(db, Form)
|
||
|
||
def get_by_name(self, name: str) -> Optional[Form]:
|
||
"""Получить форму по имени"""
|
||
return self.db.query(Form).filter(Form.name == name).first()
|
||
|
||
def get_active_forms(self, skip: int = 0, limit: int = 100) -> List[Form]:
|
||
"""Получить активные формы"""
|
||
return self.db.query(Form).filter(
|
||
Form.is_active == True,
|
||
Form.is_published == True
|
||
).offset(skip).limit(limit).all()
|
||
|
||
def get_forms_with_submissions_count(self) -> List[Dict[str, Any]]:
|
||
"""Получить формы с количеством submissions"""
|
||
from app.models.submission import Submission
|
||
|
||
results = self.db.query(
|
||
Form.id,
|
||
Form.name,
|
||
Form.description,
|
||
Form.created_at,
|
||
func.count(Submission.id).label('submissions_count')
|
||
).outerjoin(
|
||
Submission, Form.id == Submission.form_id
|
||
).group_by(
|
||
Form.id
|
||
).all()
|
||
|
||
return [
|
||
{
|
||
"id": r.id,
|
||
"name": r.name,
|
||
"description": r.description,
|
||
"created_at": r.created_at,
|
||
"submissions_count": r.submissions_count
|
||
}
|
||
for r in results
|
||
]
|
||
|
||
def clone_form(self, form_id: int, new_name: str) -> Optional[Form]:
|
||
"""Клонировать форму со всеми полями"""
|
||
original_form = self.get_by_id(form_id)
|
||
if not original_form:
|
||
return None
|
||
|
||
# Клонируем форму
|
||
cloned_form = Form(
|
||
name=new_name,
|
||
description=original_form.description,
|
||
settings=original_form.settings,
|
||
is_active=False,
|
||
is_published=False
|
||
)
|
||
self.db.add(cloned_form)
|
||
self.db.flush()
|
||
|
||
# Клонируем связи с полями
|
||
for form_field in original_form.fields:
|
||
new_form_field = FormField(
|
||
form_id=cloned_form.id,
|
||
field_id=form_field.field_id,
|
||
order=form_field.order,
|
||
is_required=form_field.is_required,
|
||
default_value=form_field.default_value,
|
||
visibility_conditions=form_field.visibility_conditions,
|
||
validation_rules_override=form_field.validation_rules_override
|
||
)
|
||
self.db.add(new_form_field)
|
||
|
||
self.db.commit()
|
||
self.db.refresh(cloned_form)
|
||
return cloned_form
|
||
|
||
|
||
class FieldRepository(BaseRepository[Field]):
|
||
def __init__(self, db: Session):
|
||
super().__init__(db, Field)
|
||
|
||
def get_by_name(self, name: str) -> Optional[Field]:
|
||
"""Получить поле по имени"""
|
||
return self.db.query(Field).filter(Field.name == name).first()
|
||
|
||
def get_or_create_field(self, field_data: Dict[str, Any]) -> Field:
|
||
"""Получить или создать поле"""
|
||
field = self.get_by_name(field_data.get("name"))
|
||
if not field:
|
||
field = self.create(field_data)
|
||
return field |