67 lines
1.6 KiB
Python
67 lines
1.6 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.iso_thread import IsoThread
|
|
|
|
|
|
async def get(db: AsyncSession, iso_thread_id: int) -> IsoThread | None:
|
|
return await db.get(IsoThread, iso_thread_id)
|
|
|
|
|
|
async def get_by_values(
|
|
db: AsyncSession,
|
|
*,
|
|
diameter_mm: float,
|
|
pitch_mm: float,
|
|
depth_mm: float | None,
|
|
) -> IsoThread | None:
|
|
depth_clause = (
|
|
IsoThread.depth_mm.is_(None)
|
|
if depth_mm is None
|
|
else IsoThread.depth_mm == depth_mm
|
|
)
|
|
result = await db.execute(
|
|
select(IsoThread).where(
|
|
IsoThread.diameter_mm == diameter_mm,
|
|
IsoThread.pitch_mm == pitch_mm,
|
|
depth_clause,
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_all(
|
|
db: AsyncSession, skip: int = 0, limit: int = 100
|
|
) -> list[IsoThread]:
|
|
result = await db.execute(
|
|
select(IsoThread).order_by(IsoThread.id).offset(skip).limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def create(
|
|
db: AsyncSession,
|
|
*,
|
|
diameter_mm: float,
|
|
pitch_mm: float,
|
|
depth_mm: float | None = None,
|
|
) -> IsoThread:
|
|
thread = IsoThread(diameter_mm=diameter_mm, pitch_mm=pitch_mm, depth_mm=depth_mm)
|
|
db.add(thread)
|
|
await db.commit()
|
|
await db.refresh(thread)
|
|
return thread
|
|
|
|
|
|
async def update(db: AsyncSession, thread: IsoThread, data: dict) -> IsoThread:
|
|
for field, value in data.items():
|
|
setattr(thread, field, value)
|
|
await db.commit()
|
|
await db.refresh(thread)
|
|
return thread
|
|
|
|
|
|
async def delete(db: AsyncSession, thread: IsoThread) -> None:
|
|
await db.delete(thread)
|
|
await db.commit()
|