62 lines
1.6 KiB
Python
62 lines
1.6 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.clamp_connection import ClampConnection
|
|
|
|
|
|
async def get(db: AsyncSession, clamp_connection_id: int) -> ClampConnection | None:
|
|
return await db.get(ClampConnection, clamp_connection_id)
|
|
|
|
|
|
async def list_all(
|
|
db: AsyncSession, skip: int = 0, limit: int = 100
|
|
) -> list[ClampConnection]:
|
|
result = await db.execute(
|
|
select(ClampConnection)
|
|
.order_by(
|
|
ClampConnection.min_mm.nulls_last(), ClampConnection.min_in.nulls_last()
|
|
)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def get_by_values(
|
|
db: AsyncSession, *, min_mm: float | None, max_mm: float | None
|
|
) -> ClampConnection | None:
|
|
result = await db.execute(
|
|
select(ClampConnection).where(
|
|
ClampConnection.min_mm == min_mm, ClampConnection.max_mm == max_mm
|
|
)
|
|
)
|
|
return result.scalars().first()
|
|
|
|
|
|
async def create(
|
|
db: AsyncSession,
|
|
*,
|
|
min_mm: float | None,
|
|
max_mm: float | None,
|
|
min_in: float | None,
|
|
max_in: float | None,
|
|
) -> ClampConnection:
|
|
cc = ClampConnection(min_mm=min_mm, max_mm=max_mm, min_in=min_in, max_in=max_in)
|
|
db.add(cc)
|
|
await db.commit()
|
|
await db.refresh(cc)
|
|
return cc
|
|
|
|
|
|
async def update(db: AsyncSession, cc: ClampConnection, data: dict) -> ClampConnection:
|
|
for field, value in data.items():
|
|
setattr(cc, field, value)
|
|
await db.commit()
|
|
await db.refresh(cc)
|
|
return cc
|
|
|
|
|
|
async def delete(db: AsyncSession, cc: ClampConnection) -> None:
|
|
await db.delete(cc)
|
|
await db.commit()
|