80 lines
2.2 KiB
Python
80 lines
2.2 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.associations import augmented_part_imported_parts
|
|
from app.models.augmented_part import AugmentedPart
|
|
from app.models.imported_part import ImportedPart
|
|
|
|
_RELATIONSHIPS = ["colour", "manufacturer", "scope_model", "part_type", "sub_part_type"]
|
|
|
|
|
|
async def get(db: AsyncSession, augmented_part_id: int) -> AugmentedPart | None:
|
|
return await db.get(AugmentedPart, augmented_part_id)
|
|
|
|
|
|
async def list_all(
|
|
db: AsyncSession, skip: int = 0, limit: int = 100
|
|
) -> list[AugmentedPart]:
|
|
result = await db.execute(
|
|
select(AugmentedPart)
|
|
.order_by(AugmentedPart.description)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def create(
|
|
db: AsyncSession,
|
|
*,
|
|
sku: str | None,
|
|
description: str,
|
|
colour_id: int,
|
|
manufacturer_id: int,
|
|
scope_model_id: int | None,
|
|
part_type_id: int,
|
|
sub_part_type_id: int | None,
|
|
) -> AugmentedPart:
|
|
ap = AugmentedPart(
|
|
sku=sku,
|
|
description=description,
|
|
colour_id=colour_id,
|
|
manufacturer_id=manufacturer_id,
|
|
scope_model_id=scope_model_id,
|
|
part_type_id=part_type_id,
|
|
sub_part_type_id=sub_part_type_id,
|
|
)
|
|
db.add(ap)
|
|
await db.commit()
|
|
await db.refresh(ap, attribute_names=_RELATIONSHIPS)
|
|
return ap
|
|
|
|
|
|
async def update(
|
|
db: AsyncSession, ap: AugmentedPart, data: dict
|
|
) -> AugmentedPart:
|
|
for field, value in data.items():
|
|
setattr(ap, field, value)
|
|
await db.commit()
|
|
await db.refresh(ap, attribute_names=_RELATIONSHIPS)
|
|
return ap
|
|
|
|
|
|
async def get_unlinked_imported_parts(db: AsyncSession) -> list[ImportedPart]:
|
|
"""ImportedParts that have no corresponding AugmentedPart link."""
|
|
result = await db.execute(
|
|
select(ImportedPart)
|
|
.where(
|
|
ImportedPart.InventoryID.not_in(
|
|
select(augmented_part_imported_parts.c.imported_part_id)
|
|
)
|
|
)
|
|
.order_by(ImportedPart.InventoryID)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def delete(db: AsyncSession, ap: AugmentedPart) -> None:
|
|
await db.delete(ap)
|
|
await db.commit()
|