122 lines
3.5 KiB
Python
122 lines
3.5 KiB
Python
from sqlalchemy import or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.attachment_point import AttachmentPoint
|
|
from app.models.fit import Fit, FitLabel
|
|
|
|
# "updated_at" is included because it's a server-side onupdate default: after an
|
|
# UPDATE, SQLAlchemy doesn't know its new value unless it's named explicitly here
|
|
# too, and access to an unrefreshed attribute would otherwise trigger a lazy load
|
|
# outside of an awaited context (MissingGreenlet) during response serialization.
|
|
_RELATIONSHIPS = ["attachment_point_a", "attachment_point_b", "notes", "updated_at"]
|
|
|
|
|
|
def _ordered(
|
|
ap_a: AttachmentPoint, ap_b: AttachmentPoint
|
|
) -> tuple[AttachmentPoint, AttachmentPoint]:
|
|
"""Canonicalize so attachment_point_a.id < attachment_point_b.id.
|
|
|
|
Keeps a single row per unordered pair of attachment points.
|
|
"""
|
|
return (ap_a, ap_b) if ap_a.id < ap_b.id else (ap_b, ap_a)
|
|
|
|
|
|
async def get(db: AsyncSession, fit_id: int) -> Fit | None:
|
|
return await db.get(Fit, fit_id)
|
|
|
|
|
|
async def get_by_attachment_points(
|
|
db: AsyncSession, ap_a: AttachmentPoint, ap_b: AttachmentPoint
|
|
) -> Fit | None:
|
|
first, second = _ordered(ap_a, ap_b)
|
|
result = await db.execute(
|
|
select(Fit).where(
|
|
Fit.attachment_point_a_id == first.id,
|
|
Fit.attachment_point_b_id == second.id,
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_for_part(db: AsyncSession, augmented_part_id: int) -> list[Fit]:
|
|
result = await db.execute(
|
|
select(Fit)
|
|
.where(
|
|
or_(
|
|
Fit.augmented_part_a_id == augmented_part_id,
|
|
Fit.augmented_part_b_id == augmented_part_id,
|
|
)
|
|
)
|
|
.order_by(Fit.id)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def list_for_parts(
|
|
db: AsyncSession,
|
|
part_ids: list[int],
|
|
exclude_label: FitLabel | None = None,
|
|
) -> list[Fit]:
|
|
"""All fits touching any of the given augmented parts, in one query.
|
|
|
|
Pass ``exclude_label`` to drop, e.g., confirmed non-fits (``FitLabel.NONE``).
|
|
"""
|
|
if not part_ids:
|
|
return []
|
|
q = select(Fit).where(
|
|
or_(
|
|
Fit.augmented_part_a_id.in_(part_ids),
|
|
Fit.augmented_part_b_id.in_(part_ids),
|
|
)
|
|
)
|
|
if exclude_label is not None:
|
|
q = q.where(Fit.label != exclude_label)
|
|
result = await db.execute(q.order_by(Fit.id))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def list_all(
|
|
db: AsyncSession,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
label: FitLabel | None = None,
|
|
) -> list[Fit]:
|
|
q = select(Fit)
|
|
if label is not None:
|
|
q = q.where(Fit.label == label)
|
|
result = await db.execute(q.order_by(Fit.id).offset(skip).limit(limit))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def create(
|
|
db: AsyncSession,
|
|
*,
|
|
ap_a: AttachmentPoint,
|
|
ap_b: AttachmentPoint,
|
|
label: FitLabel,
|
|
) -> Fit:
|
|
first, second = _ordered(ap_a, ap_b)
|
|
fit = Fit(
|
|
attachment_point_a_id=first.id,
|
|
attachment_point_b_id=second.id,
|
|
augmented_part_a_id=first.augmented_part_id,
|
|
augmented_part_b_id=second.augmented_part_id,
|
|
label=label,
|
|
)
|
|
db.add(fit)
|
|
await db.commit()
|
|
await db.refresh(fit, attribute_names=_RELATIONSHIPS)
|
|
return fit
|
|
|
|
|
|
async def set_label(db: AsyncSession, fit: Fit, label: FitLabel) -> Fit:
|
|
fit.label = label
|
|
await db.commit()
|
|
await db.refresh(fit, attribute_names=_RELATIONSHIPS)
|
|
return fit
|
|
|
|
|
|
async def delete(db: AsyncSession, fit: Fit) -> None:
|
|
await db.delete(fit)
|
|
await db.commit()
|