71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.scope_info import ScopeInfo
|
|
|
|
|
|
async def get(db: AsyncSession, scope_info_id: int) -> ScopeInfo | None:
|
|
return await db.get(ScopeInfo, scope_info_id)
|
|
|
|
|
|
async def get_by_augmented_part(
|
|
db: AsyncSession, augmented_part_id: int
|
|
) -> ScopeInfo | None:
|
|
result = await db.execute(
|
|
select(ScopeInfo).where(ScopeInfo.augmented_part_id == augmented_part_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_by_augmented_parts(
|
|
db: AsyncSession, augmented_part_ids: list[int]
|
|
) -> list[ScopeInfo]:
|
|
"""ScopeInfo rows for the given augmented parts, fetched in one query."""
|
|
if not augmented_part_ids:
|
|
return []
|
|
result = await db.execute(
|
|
select(ScopeInfo).where(ScopeInfo.augmented_part_id.in_(augmented_part_ids))
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def create(
|
|
db: AsyncSession,
|
|
*,
|
|
augmented_part_id: int,
|
|
magnification_min: float | None = None,
|
|
magnification_max: float | None = None,
|
|
model_suffix: str | None = None,
|
|
objective_lens_diameter_mm: float | None = None,
|
|
objective_thread_pitch: str | None = None,
|
|
) -> ScopeInfo:
|
|
info = ScopeInfo(
|
|
augmented_part_id=augmented_part_id,
|
|
magnification_min=magnification_min,
|
|
magnification_max=magnification_max,
|
|
model_suffix=model_suffix,
|
|
objective_lens_diameter_mm=objective_lens_diameter_mm,
|
|
objective_thread_pitch=objective_thread_pitch,
|
|
)
|
|
db.add(info)
|
|
await db.commit()
|
|
await db.refresh(info)
|
|
return info
|
|
|
|
|
|
async def upsert(db: AsyncSession, *, augmented_part_id: int, **fields) -> ScopeInfo:
|
|
"""Create the ScopeInfo for an AugmentedPart, or update it if it exists."""
|
|
info = await get_by_augmented_part(db, augmented_part_id)
|
|
if info is None:
|
|
return await create(db, augmented_part_id=augmented_part_id, **fields)
|
|
for field, value in fields.items():
|
|
setattr(info, field, value)
|
|
await db.commit()
|
|
await db.refresh(info)
|
|
return info
|
|
|
|
|
|
async def delete(db: AsyncSession, info: ScopeInfo) -> None:
|
|
await db.delete(info)
|
|
await db.commit()
|