60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
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
|