51 lines
1.3 KiB
Python
51 lines
1.3 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.attachment_point_name import AttachmentPointName
|
|
|
|
|
|
async def get(db: AsyncSession, apn_id: int) -> AttachmentPointName | None:
|
|
return await db.get(AttachmentPointName, apn_id)
|
|
|
|
|
|
async def get_by_name(db: AsyncSession, name: str) -> AttachmentPointName | None:
|
|
result = await db.execute(
|
|
select(AttachmentPointName).where(AttachmentPointName.name == name)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_all(
|
|
db: AsyncSession, skip: int = 0, limit: int = 100
|
|
) -> list[AttachmentPointName]:
|
|
result = await db.execute(
|
|
select(AttachmentPointName)
|
|
.order_by(AttachmentPointName.name)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def create(db: AsyncSession, *, name: str) -> AttachmentPointName:
|
|
apn = AttachmentPointName(name=name)
|
|
db.add(apn)
|
|
await db.commit()
|
|
await db.refresh(apn)
|
|
return apn
|
|
|
|
|
|
async def update(
|
|
db: AsyncSession, apn: AttachmentPointName, data: dict
|
|
) -> AttachmentPointName:
|
|
for field, value in data.items():
|
|
setattr(apn, field, value)
|
|
await db.commit()
|
|
await db.refresh(apn)
|
|
return apn
|
|
|
|
|
|
async def delete(db: AsyncSession, apn: AttachmentPointName) -> None:
|
|
await db.delete(apn)
|
|
await db.commit()
|