zk
66 lines
3.4 KiB
Python
66 lines
3.4 KiB
Python
"""简历统一查询模块
|
|
|
|
提供简历主表 + 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
|