Cheng Zhou
39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select, update as sa_update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.etmf import EtmfNode
|
|
|
|
|
|
async def get_node(db: AsyncSession, node_id: uuid.UUID) -> EtmfNode | None:
|
|
result = await db.execute(select(EtmfNode).where(EtmfNode.id == node_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_nodes_by_study(db: AsyncSession, study_id: uuid.UUID, *, include_inactive: bool = True) -> Sequence[EtmfNode]:
|
|
stmt = select(EtmfNode).where(EtmfNode.study_id == study_id)
|
|
if not include_inactive:
|
|
stmt = stmt.where(EtmfNode.is_active.is_(True))
|
|
stmt = stmt.order_by(EtmfNode.sort_order.asc(), EtmfNode.code.asc(), EtmfNode.name.asc())
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def create_node(db: AsyncSession, node: EtmfNode, *, commit: bool = True) -> EtmfNode:
|
|
db.add(node)
|
|
if commit:
|
|
await db.commit()
|
|
await db.refresh(node)
|
|
return node
|
|
|
|
|
|
async def update_node(db: AsyncSession, node_id: uuid.UUID, values: dict, *, commit: bool = True) -> None:
|
|
if values:
|
|
await db.execute(sa_update(EtmfNode).where(EtmfNode.id == node_id).values(**values))
|
|
if commit:
|
|
await db.commit()
|