抽象简历查询

app/services/resume_loader.py

@@ -0,0 +1,65 @@
+"""简历统一查询模块
+
+提供简历主表 + 5张子表的统一查询能力，返回脱离 session 的 ResumeDetail dataclass。
+各 Service 统一复用，避免重复查询逻辑。
+"""
+
+from dataclasses import dataclass, field
+
+from sqlalchemy import select, desc
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.user_resume import UserResume
+from app.models.user_resume_education import UserResumeEducation
+from app.models.user_resume_work import UserResumeWork
+from app.models.user_resume_internship import UserResumeInternship
+from app.models.user_resume_project import UserResumeProject
+from app.models.user_resume_competition import UserResumeCompetition
+
+
+@dataclass
+class ResumeDetail:
+    """简历完整数据，主表 + 5张子表"""
+    resume: UserResume
+    education: list[UserResumeEducation] = field(default_factory=list)
+    work: list[UserResumeWork] = field(default_factory=list)
+    internship: list[UserResumeInternship] = field(default_factory=list)
+    project: list[UserResumeProject] = field(default_factory=list)
+    competition: list[UserResumeCompetition] = field(default_factory=list)
+
+
+async def load_resume_detail(session: AsyncSession, resume_id: int, user_id: int) -> ResumeDetail:
+    """按ID查简历主表（校验归属）+ 5张子表，返回 ResumeDetail"""
+    result = await session.execute(select(UserResume).where(UserResume.id == resume_id, UserResume.user_id == user_id))
+    resume = result.scalar_one_or_none()
+    if not resume:
+        raise ValueError("简历不存在")
+    edu, work, intern, proj, comp = await _load_sub_tables(session, resume_id)
+    return ResumeDetail(resume=resume, education=edu, work=work, internship=intern, project=proj, competition=comp)
+
+
+async def load_default_resume_detail(session: AsyncSession, user_id: int) -> ResumeDetail:
+    """自动选默认简历（先默认再最新）+ 5张子表，返回 ResumeDetail"""
+    result = await session.execute(
+        select(UserResume).where(UserResume.user_id == user_id, UserResume.is_default == 1)
+        .order_by(desc(UserResume.update_time)).limit(1))
+    resume = result.scalar_one_or_none()
+    if not resume:
+        result = await session.execute(
+            select(UserResume).where(UserResume.user_id == user_id)
+            .order_by(desc(UserResume.update_time)).limit(1))
+        resume = result.scalar_one_or_none()
+    if not resume:
+        raise ValueError("请先创建简历")
+    edu, work, intern, proj, comp = await _load_sub_tables(session, resume.id)
+    return ResumeDetail(resume=resume, education=edu, work=work, internship=intern, project=proj, competition=comp)
+
+
+async def _load_sub_tables(session: AsyncSession, resume_id: int):
+    """查询简历5张子表"""
+    edu = (await session.execute(select(UserResumeEducation).where(UserResumeEducation.resume_id == resume_id))).scalars().all()
+    work = (await session.execute(select(UserResumeWork).where(UserResumeWork.resume_id == resume_id))).scalars().all()
+    intern = (await session.execute(select(UserResumeInternship).where(UserResumeInternship.resume_id == resume_id))).scalars().all()
+    proj = (await session.execute(select(UserResumeProject).where(UserResumeProject.resume_id == resume_id))).scalars().all()
+    comp = (await session.execute(select(UserResumeCompetition).where(UserResumeCompetition.resume_id == resume_id))).scalars().all()
+    return edu, work, intern, proj, comp