from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from models.library import LibraryFile from schemas.library import LibraryFileRequest, LibraryFileResponse async def list_files(db: AsyncSession) -> list[LibraryFileResponse]: stmt = select(LibraryFile).order_by(LibraryFile.create_time.desc()) files = (await db.execute(stmt)).scalars().all() return [ LibraryFileResponse( id=file.id, subject=file.subject, module=file.module, name=file.name, type=file.type, size=file.size, content=file.content, uploadTime=file.create_time.strftime("%Y-%m-%d %H:%M:%S"), ) for file in files ] async def add_file(db: AsyncSession, data: LibraryFileRequest) -> bool: """ 创建题库文件 """ file = LibraryFile( subject=data.subject, module=data.module, name=data.name, size=data.size, type=data.type, content=data.content, ) db.add(file) await db.commit() return True async def edit_file(db: AsyncSession, file_id: int, data: LibraryFileRequest) -> bool: """ 编辑题库文件 """ result = await db.execute( select(LibraryFile).where(LibraryFile.id == file_id) ) file = result.scalar_one_or_none() if not file: return False file.subject = data.subject file.module = data.module file.name = data.name file.size = data.size file.type = data.type file.content = data.content await db.commit() return True async def get_file(db: AsyncSession, file_id: int) -> LibraryFileResponse: stmt = (select(LibraryFile).where(LibraryFile.id == file_id)) result = await db.execute(stmt) file = result.scalar_one_or_none() return LibraryFileResponse( id=file.id, subject=file.subject, module=file.module, name=file.name, type=file.type, size=file.size, content=file.content, uploadTime=file.create_time.strftime("%Y-%m-%d %H:%M:%S"), )