e5a725fc91
- database.py: engine/SessionLocal 改为延迟创建，避免 import 时副作用， 并提供 dispose_engine() 支持配置热重载 - dao/uploaded_file_dao.py: 新增 count() 方法 - upload_service.py: list_files() 改用 dao.count() 返回数据库总条数， 修复之前返回当前页条目数导致分页 total 语义错误的问题
48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
"""UploadedFile 的 DAO。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..models.uploaded_file import UploadedFile
|
|
|
|
|
|
class UploadedFileDAO:
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
|
|
def create(self, file: UploadedFile) -> UploadedFile:
|
|
self.db.add(file)
|
|
self.db.commit()
|
|
self.db.refresh(file)
|
|
return file
|
|
|
|
def get_by_id(self, file_id: int) -> UploadedFile | None:
|
|
return self.db.get(UploadedFile, file_id)
|
|
|
|
def get_by_sha256(self, sha256: str) -> UploadedFile | None:
|
|
stmt = select(UploadedFile).where(UploadedFile.sha256 == sha256).limit(1)
|
|
return self.db.scalars(stmt).first()
|
|
|
|
def count(self) -> int:
|
|
"""返回数据库中文件总条数。"""
|
|
return self.db.scalar(select(func.count()).select_from(UploadedFile)) or 0
|
|
|
|
def list(self, limit: int = 100, offset: int = 0) -> list[UploadedFile]:
|
|
stmt = (
|
|
select(UploadedFile)
|
|
.order_by(UploadedFile.uploaded_at.desc())
|
|
.limit(limit)
|
|
.offset(offset)
|
|
)
|
|
return list(self.db.scalars(stmt).all())
|
|
|
|
def delete(self, file_id: int) -> bool:
|
|
file = self.get_by_id(file_id)
|
|
if file is None:
|
|
return False
|
|
self.db.delete(file)
|
|
self.db.commit()
|
|
return True
|