Cheng Zhou
57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.site import Site
|
|
from app.schemas.site import SiteCreate, SiteUpdate
|
|
|
|
|
|
async def create_site(db: AsyncSession, study_id: uuid.UUID, site_in: SiteCreate) -> Site:
|
|
site = Site(
|
|
study_id=study_id,
|
|
name=site_in.name,
|
|
city=site_in.city,
|
|
pi_name=site_in.pi_name,
|
|
contact=site_in.contact,
|
|
is_active=site_in.is_active,
|
|
)
|
|
db.add(site)
|
|
await db.commit()
|
|
await db.refresh(site)
|
|
return site
|
|
|
|
|
|
async def get_site(db: AsyncSession, site_id: uuid.UUID) -> Site | None:
|
|
result = await db.execute(select(Site).where(Site.id == site_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def update_site(db: AsyncSession, site: Site, site_in: SiteUpdate) -> Site:
|
|
update_data = site_in.model_dump(exclude_unset=True)
|
|
if update_data:
|
|
await db.execute(
|
|
update(Site)
|
|
.where(Site.id == site.id)
|
|
.values(**update_data)
|
|
)
|
|
await db.commit()
|
|
await db.refresh(site)
|
|
return site
|
|
|
|
|
|
async def list_by_study(db: AsyncSession, study_id: uuid.UUID, skip: int = 0, limit: int = 100) -> Sequence[Site]:
|
|
result = await db.execute(
|
|
select(Site).where(Site.study_id == study_id).offset(skip).limit(limit)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def get_sites_by_ids(db: AsyncSession, ids: set[uuid.UUID]) -> dict[uuid.UUID, Site]:
|
|
if not ids:
|
|
return {}
|
|
result = await db.execute(select(Site).where(Site.id.in_(ids)))
|
|
sites = result.scalars().all()
|
|
return {s.id: s for s in sites}
|