69 lines
1.9 KiB
Python
69 lines
1.9 KiB
Python
import enum
|
|
import itertools
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.db.models.form_type import FormType
|
|
|
|
|
|
class DirectionEnum(str, enum.Enum):
|
|
SUPPORT = "Support"
|
|
DEVELOPMENT = "Development"
|
|
|
|
ACCEPTABLE_SHEETS = set(itertools.chain(*FormType.SHEETS_BY_FORM_TYPE.values()))
|
|
ACCEPTABLE_SECTIONS = set(itertools.chain(*FormType.SECTIONS_BY_FORM_TYPE.values()))
|
|
|
|
|
|
|
|
class SheetValidationError(Exception):
|
|
|
|
def __init__(self, message, *args):
|
|
super().__init__(*args)
|
|
self.message = message
|
|
|
|
|
|
class SheetRepository:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def get(
|
|
self,
|
|
form_id: int,
|
|
sheet: str,
|
|
direction: str,
|
|
sections: list[str] | None = None,
|
|
) -> list[tuple]:
|
|
if any(
|
|
[
|
|
not isinstance(form_id, int),
|
|
sheet not in ACCEPTABLE_SHEETS,
|
|
direction not in [s.value for s in DirectionEnum],
|
|
sections and (set(sections) - ACCEPTABLE_SECTIONS),
|
|
]
|
|
):
|
|
raise SheetValidationError("Некорректные данные")
|
|
|
|
sections_str = 'NULL'
|
|
if sections:
|
|
sections_str = ",".join([f"'{section}'" for section in sections])
|
|
sections_str = f"ARRAY[{sections_str}]"
|
|
if direction:
|
|
func_query = "v3.v_form_view(%s, '%s', %s, '%s')" % (
|
|
form_id, sheet, sections_str, direction,
|
|
)
|
|
else:
|
|
func_query = "v3.v_form_view(%s, '%s', %s)" % (
|
|
form_id, sheet, sections_str,
|
|
)
|
|
result = (
|
|
await self.db.execute(
|
|
text(
|
|
f"""
|
|
SELECT row_type, depth, sort_order, data
|
|
FROM {func_query}
|
|
"""
|
|
)
|
|
)
|
|
).all()
|
|
return result |