56 lines
1.4 KiB
Python
56 lines
1.4 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.static_connection import StaticConnection
|
|
|
|
|
|
async def get(db: AsyncSession, static_connection_id: int) -> StaticConnection | None:
|
|
return await db.get(StaticConnection, static_connection_id)
|
|
|
|
|
|
async def get_by_od_mm(db: AsyncSession, od_mm: float) -> StaticConnection | None:
|
|
result = await db.execute(
|
|
select(StaticConnection).where(StaticConnection.od_mm == od_mm)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_all(
|
|
db: AsyncSession, skip: int = 0, limit: int = 100
|
|
) -> list[StaticConnection]:
|
|
result = await db.execute(
|
|
select(StaticConnection)
|
|
.order_by(StaticConnection.od_mm)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def create(
|
|
db: AsyncSession,
|
|
*,
|
|
od_mm: float,
|
|
od_in: float,
|
|
) -> StaticConnection:
|
|
sc = StaticConnection(od_mm=od_mm, od_in=od_in)
|
|
db.add(sc)
|
|
await db.commit()
|
|
await db.refresh(sc)
|
|
return sc
|
|
|
|
|
|
async def update(
|
|
db: AsyncSession, sc: StaticConnection, data: dict
|
|
) -> StaticConnection:
|
|
for field, value in data.items():
|
|
setattr(sc, field, value)
|
|
await db.commit()
|
|
await db.refresh(sc)
|
|
return sc
|
|
|
|
|
|
async def delete(db: AsyncSession, sc: StaticConnection) -> None:
|
|
await db.delete(sc)
|
|
await db.commit()
|