96 lines
2.5 KiB
Python
96 lines
2.5 KiB
Python
from sqlalchemy import select, func
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from models.family import Records, Comments, Likes
|
|
from schemas.record import RecordCreate, RecordUpdate, RecordResponse, RecordPageResponse
|
|
|
|
|
|
def get_all_records_by_page(db: Session, user_id: int, page: int = 1, size: int = 20) -> RecordPageResponse:
|
|
base_query = (
|
|
select(Records)
|
|
.options(
|
|
selectinload(Records.user),
|
|
selectinload(Records.comments).selectinload(Comments.user),
|
|
)
|
|
.order_by(Records.create_time.desc())
|
|
)
|
|
|
|
# 总数
|
|
total = db.scalar(
|
|
select(func.count()).select_from(base_query.subquery())
|
|
)
|
|
|
|
# 分页数据
|
|
records = (
|
|
db.execute(base_query.offset((page - 1) * size).limit(size))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
|
|
# 批量判断当前用户点赞状态
|
|
liked_record_ids: set[int] = set()
|
|
|
|
if user_id and records:
|
|
liked_record_ids = set(
|
|
db.scalars(
|
|
select(Likes.record_id).where(
|
|
Likes.record_id.in_([r.id for r in records]),
|
|
Likes.user_id == user_id,
|
|
)
|
|
).all()
|
|
)
|
|
|
|
# 组装响应
|
|
response_records = []
|
|
for r in records:
|
|
resp = RecordResponse.model_validate(r, from_attributes=True)
|
|
resp.liked = r.id in liked_record_ids
|
|
response_records.append(resp)
|
|
|
|
return RecordPageResponse(
|
|
records=response_records,
|
|
total=total,
|
|
page=page,
|
|
size=size,
|
|
pages=(total + size - 1) // size,
|
|
)
|
|
|
|
|
|
def create_record(db: Session, record_dto: RecordCreate) -> bool:
|
|
data = record_dto.model_dump()
|
|
record = Records(**data)
|
|
db.add(record)
|
|
db.commit()
|
|
db.refresh(record)
|
|
|
|
return True
|
|
|
|
|
|
def get_record_by_id(db: Session, record_id: int) -> Records | None:
|
|
result = db.execute(select(Records).where(Records.id == record_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
def update_record(db: Session, record_id: int, record_dto: RecordUpdate) -> bool:
|
|
record = get_record_by_id(db, record_id)
|
|
if not record:
|
|
return False
|
|
|
|
for field, value in record_dto.model_dump(exclude_unset=True).items():
|
|
setattr(record, field, value)
|
|
db.commit()
|
|
db.refresh(record)
|
|
|
|
return True
|
|
|
|
|
|
def delete_record(db: Session, record_id: int) -> bool:
|
|
record = get_record_by_id(db, record_id)
|
|
if not record:
|
|
return False
|
|
|
|
db.delete(record)
|
|
db.commit()
|
|
|
|
return True
|