feat: 初始化工程

service/record_service.py

@@ -0,0 +1,55 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from models.patch import Records
+from schemas.patch import RecordCreate, RecordResponse, RecordUpdate
+
+
+def create_record(db: Session, data: RecordCreate) -> bool:
+    record = Records(**data.model_dump())
+    db.add(record)
+    db.commit()
+    db.refresh(record)
+
+    return True
+
+
+def get_record(db: Session, record_id: int) -> RecordResponse | None:
+    result = db.execute(select(Records).where(Records.id == record_id))
+    record = result.scalar_one_or_none()
+
+    return RecordResponse.from_orm(record) if record else None
+
+
+def get_records_by_crop(db: Session, crop_id: int) -> list[RecordResponse]:
+    result = db.execute(
+        select(Records)
+        .where(Records.crop_id == crop_id)
+        .order_by(Records.date.desc())
+    )
+    return [RecordResponse.from_orm(r) for r in result.scalars().all()]
+
+
+def update_record(db: Session, record_id: int, data: RecordUpdate) -> bool:
+    result = db.execute(select(Records).where(Records.id == record_id))
+    record = result.scalar_one_or_none()
+    if not record:
+        return False
+
+    for k, v in data.model_dump(exclude_unset=True).items():
+        setattr(record, k, v)
+    db.commit()
+    db.refresh(record)
+
+    return True
+
+
+def delete_record(db: Session, record_id: int) -> bool:
+    result = db.execute(select(Records).where(Records.id == record_id))
+    record = result.scalar_one_or_none()
+    if record:
+        db.delete(record)
+        db.commit()
+        return True
+
+    return False