56 lines
1.5 KiB
Python
56 lines
1.5 KiB
Python
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
|