feat: 初始化工程

service/record_service.py

@@ -0,0 +1,59 @@
+from sqlalchemy import select, desc
+from sqlalchemy.orm import Session, selectinload
+
+from models.family import Records, Comments
+from schemas.record import RecordCreate, RecordUpdate, RecordResponse
+
+
+def get_all_records(db: Session, skip: int = 0, limit: int = 20) -> list[RecordResponse]:
+    result = db.execute(
+        select(Records)
+        .options(
+            selectinload(Records.user),
+            selectinload(Records.comments).selectinload(Comments.user),
+        )
+        .order_by(desc(Records.create_time))
+        .offset(skip)
+        .limit(limit)
+    )
+    records = result.scalars().all()
+    return [RecordResponse.model_validate(r) for r in records]
+
+
+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