項目實踐FastAPI Elasticsearch 職位數據全量同步FastAPI Tortoise-ORM Elasticsearch 實戰職位數據全量同步到 ES 的設計意識與踩坑復盤一、前言介紹1.1 功能定位1.2 數據模型總覽1.3 同步流程總覽二、環境準備2.1 依賴與 ES 客戶端2.2 索引與運行環境2.3 路由與生命周期掛載三、知識點講解3.1 ORM 與 ES 的數據形態差異寬表思想3.2 IntEnumField 與 JSONField 的存儲語義3.3 text / keyword / date 三類字段選型3.4 異步客戶端單例與依賴注入3.5 批量寫入與冪等_id async_bulk3.6 N1 查詢的批量化解法四、代碼邏輯拆解4.1 ES 客戶端單例與依賴橋接4.2 值序列化工具 _to_es_value4.3 寬表文檔拼裝 _build_job_document4.4 創建索引mapping 設計4.5 全量同步 insert-data-v24.6 職位寫入接口 saveJobFastAPI Tortoise-ORM Elasticsearch 實戰職位數據全量同步到 ES 的設計意識與踩坑復盤一、前言介紹1.1 功能定位本文只講兩件事的設計意識其一職位領域模型與寫入接口怎么落地其二如何把 MySQL 里的職位數據批量、可靠地同步進 ES。1.2 數據模型總覽三個實體之間的關系是同步邏輯的基礎Job職位表 t_job └── enterprise_id : IntField ──┐ 手動整型關聯不做外鍵級聯 ↓ Enterprise企業主表 ── 1:1 ── EnterpriseInfo企業工商信息 └── industry : ForeignKeyField → Industry行業 Industry行業字典/ City城市字典要點Job與企業之間用的是enterprise_id普通整型字段而非ForeignKeyField這意味著同步時要手動按 ID 取企業且不會因為企業被刪而級聯掉職位。1.3 同步流程總覽啟動期lifespan 里初始化 ES 異步客戶端單例 ↓ 建索引POST /es-data/create-index-v2 → 聲明 mapping字段類型 IK 分詞 ↓ 全量同步POST /es-data/insert-data-v2 Job.all() → 收集 enterprise_id批量取 Enterprise / EnterpriseInfo 進內存字典 → 每條職位拼成寬表文檔 → async_bulk 批量寫入_idjob.id 保證冪等二、環境準備2.1 依賴與 ES 客戶端同步能力依賴官方elasticsearch異步客戶端。連接信息從環境變量讀取缺省回落到本機ES_HOSTos.getenv(ES_HOST,http://localhost:9200)es_client:AsyncElasticsearch|NoneNoneos.getenv(ES_HOST, ...)優先取環境變量便于不同環境切換 ES 地址es_client用模塊級None占位后續做單例整個進程只建一條連接。2.2 索引與運行環境ES 必須安裝IK 分詞插件ik_max_word否則analyzer: ik_max_word建索引會報錯本地開發可用單分片零副本number_of_shards: 1, number_of_replicas: 0職位索引取名boss_job_index_v2刻意與舊版boss_job_index分開便于對照學習。2.3 路由與生命周期掛載ES 客戶端在應用啟動期就初始化避免首次請求時再去建連asynccontextmanagerasyncdeflifespan(app:FastAPI):awaitTortoise.init(configTORTOISE_ORM,_enable_global_fallbackTrue)...awaitget_es_client()# 啟動即建立 ES 連接單例yieldawaitTortoise.close_connections()lifespan是 FastAPI 的啟動/關閉鉤子在yield之前做的都是啟動準備之后是優雅關閉這里只關了 TortoiseES 客戶端關閉函數雖已備好但并未在此顯式調用見問題排查 5.6。三、知識點講解3.1 ORM 與 ES 的數據形態差異寬表思想MySQL 里職位、企業、工商信息、行業分表存儲靠關聯還原。ES 是文檔型存儲更適合把一次搜索要展示的所有字段拍平成一條文檔避免搜索時再回查多表。這就是同步腳本把四張表拼成一份文檔的根本動機。3.2 IntEnumField 與 JSONField 的存儲語義statusfields.IntEnumField(enum_typeJobStatus,description0:草稿,1:招聘中...)department_idfields.IntEnumField(enum_typeDeptType,description所屬部門)job_tagsfields.JSONField(defaultlist,description職位標簽示例[五險一金,年終獎])IntEnumField數據庫存的是整數但 ORM 層自動轉成枚舉對象寫代碼用JobStatus.RECRUITING比裸數字更安全JSONField數據庫列里直接存 JSON 數組job_tags變成[五險一金,年終獎]進 ES 時映射成keyword多值字段。3.3 text / keyword / date 三類字段選型這是 mapping 設計的核心判斷text ik_max_word職位名、公司名、行業、經營范圍——需要中文全文檢索keyword薪資、城市、標簽、統一社會信用代碼——用于精確匹配 / 聚合 / 篩選不分詞integer / long狀態、部門、企業 ID——用于數值范圍與等值篩選date發布時間、認證時間——用于時間區間查詢寫入時用 ISO 字符串。3.4 異步客戶端單例與依賴注入asyncdefget_es_client()-AsyncElasticsearch:globales_clientifes_clientisNone:es_clientAsyncElasticsearch(ES_HOST)logger.info(ES 客戶端初始化成功)returnes_clientglobal es_client允許函數內修改模塊級變量二次進入直接返回已建好的實例進程內復用一條連接避免每次請求都握手接口層通過Depends(es_client_depend)拿到它與鑒權依賴寫法一致。3.5 批量寫入與冪等_id async_bulkactions.append({_index:BOSS_JOB_INDEX_NAME_V2,_id:str(job.id),_source:document,})awaitasync_bulk(clientes_client,actionsactions,raise_on_errorFalse)_idstr(job.id)把職位主鍵設為文檔 ID重復同步時覆蓋舊文檔而非新增天然冪等async_bulk一次性批量提交遠比逐條index()快raise_on_errorFalse單條失敗不中斷整批返回值里能拿到錯誤列表做統計。3.6 N1 查詢的批量化解法職位有 N 條若每條都現場查企業、查工商信息就是2N次查詢N1 的變體。正確做法是先收集所有enterprise_id兩批查完建字典內存里 O(1) 關聯enterprise_idslist({job.enterprise_idforjobinjobsifjob.enterprise_idisnotNone})enterprisesawaitEnterprise.filter(id__inenterprise_ids).prefetch_related(city)enterprise_map{e.id:eforeinenterprises}集合去重避免重復查詢prefetch_related(city)一次性把城市關聯出來后面取enterprise.city.name不再發 SQLenterprise_map以企業 ID 為鍵的字典拼文檔時直接enterprise_map.get(job.enterprise_id)。四、代碼邏輯拆解4.1 ES 客戶端單例與依賴橋接asyncdefes_client_depend()-AsyncElasticsearch:returnawaitget_es_client()這是 FastAPI 依賴函數作用是在路由簽名里優雅注入 ES 客戶端內部直接復用上面講的單例保證全鏈路同一連接。4.2 值序列化工具 _to_es_valueES 只認 JSON 友好的類型ORM 里的datetime、date、枚舉要先轉換def_to_es_value(value:Any)-Any:ifvalueisNone:returnNoneifisinstance(value,datetime):returnvalue.isoformat()ifisinstance(value,date):returnvalue.isoformat()ifisinstance(value,Enum):returnvalue.valuereturnvalue第 1 行空值原樣返回Nonemapping 里對應字段可空第 2–3 行datetime/date轉 ISO 字符串ESdate字段才能識別第 4 行Enum轉成.value一般是 int否則 ES 寫入枚舉對象會序列化失敗最后一行其余類型str、int、list、dict、None原樣透傳。4.3 寬表文檔拼裝 _build_job_document這是同步的靈魂——把職位、企業、工商、行業壓成一條扁平文檔citygetattr(enterprise,city,None)ifenterpriseelseNoneindustrygetattr(enterprise_info,industry,None)ifenterprise_infoelseNonereturn{job_id:job.id,job_name:job.job_name,min_salary:job.min_salary,max_salary:job.max_salary,job_tags:job.job_tagsor[],status:_to_es_value(job.status),publish_time:_to_es_value(job.publish_time),enterprise_name:enterprise.enterprise_nameifenterpriseelseNone,enterprise_account_status:_to_es_value(enterprise.account_status)ifenterpriseelseNone,enterpriseInfo_company_scale:_to_es_value(enterprise_info.company_scale)ifenterprise_infoelseNone,industry_id:industry.idifindustryelseNone,industry_name:industry.nameifindustryelseNone,}getattr(enterprise, city, None)防御式取值企業對象沒有city屬性時不拋異常job_tags or []標簽為空時給空數組避免 ES 收到None與keyword多值類型沖突if enterprise else None企業缺失時整組企業字段填None保證單條臟數據不會打斷整批字段名帶enterpriseInfo_前綴是為了和 mapping 一一對應也能直觀區分屬于企業詳情凡是status、publish_time、company_scale等枚舉/時間字段一律過_to_es_value。4.4 創建索引mapping 設計建索引時聲明每個字段類型與中文分詞器這些是經舊版踩坑后補全的job_name:{type:text,analyzer:ik_max_word},enterprise_name:{type:text,analyzer:ik_max_word},work_location:{type:keyword},min_salary:{type:keyword},status:{type:integer},publish_time:{type:date},enterpriseInfo_business_scope:{type:text,analyzer:ik_max_word},職位名、公司名、經營范圍用text ik_max_word支持中文全文檢索城市、薪資用keyword因為要做精確篩選與聚合不能分詞status用integer和IntEnumField存的整數對齊enterpriseInfo_business_scope舊版 mapping 漏聲明卻仍在寫入導致該字段無法被檢索這里顯式補回text類型建索引前先indices.exists判斷已存在則直接返回避免重復創建報錯。4.5 全量同步 insert-data-v2核心流程先校驗索引存在再批量取數再組裝 bulk最后一次性寫入。ifnotawaites_client.indices.exists(indexBOSS_JOB_INDEX_NAME_V2):return{code:0,message:f索引{BOSS_JOB_INDEX_NAME_V2}不存在請先調用 /es-data/create-index-v2}jobsawaitJob.all()ifnotjobs:return{code:1,message:沒有可同步的職位,data:{success:0,skip:0}}第一步先確認目標索引已建否則 bulk 時才報錯浪費前面查庫的開銷Job.all()一次取出全部職位空集合提前返回避免后面空跑。forjobinjobs:enterpriseenterprise_map.get(job.enterprise_id)enterprise_infoenterprise_info_map.get(job.enterprise_id)ifenterpriseisNone:skip_count1logger.warning(f同步跳過職位 id{job.id}關聯企業 id{job.enterprise_id}不存在)continueifenterprise_infoisNone:logger.warning(f職位 id{job.id}無企業詳情將寫入空的 enterpriseInfo_* 字段)document_build_job_document(job,enterprise,enterprise_info)actions.append({_index:BOSS_JOB_INDEX_NAME_V2,_id:str(job.id),_source:document})enterprise_map.get(...)內存字典取值O(1)無額外 SQL企業主數據缺失則continue跳過寬表缺核心信息寫入也搜不到意義不大并記日志企業詳情缺失允許繼續只是enterpriseInfo_*字段為None每條文檔帶_idstr(job.id)這是冪等覆蓋的關鍵全部 action 先攢進列表最后統一async_bulk。success_count,errorsawaitasync_bulk(clientes_client,actionsactions,raise_on_errorFalse,)error_countlen(errors)ifisinstance(errors,list)else0return{code:1,message:數據同步完成,data:{index:BOSS_JOB_INDEX_NAME_V2,job_total:len(jobs),success:success_count,skip:skip_count,error:error_count},}success_count是成功條數errors是失敗詳情列表raise_on_errorFalse讓部分失敗不影響整體返回里把成功/跳過/失敗三類數量都帶上便于對賬。4.6 職位寫入接口 saveJob職位落庫時企業 ID 不是前端傳的而是從登錄的招聘團隊反查出來避免越權掛到別家企業名下staticmethodasyncdefsaveJob(job:JobCreateRequest,time_id:int):timeawaitRecruitTeam.filter(idtime_id).first()awaitJob.create(**job.dict(),enterprise_idtime.enterprise_id,recruit_team_idtime_id,publish_timenow(),)time_id來自 JWT 鑒權依賴get_job_info代表當前招聘團隊RecruitTeam.filter(idtime_id).first()查出團隊取其enterprise_id**job.dict()把請求體字段展開成建表參數省去逐字段賦值enterprise_idtime.enterprise_id企業歸屬由服務端決定前端無法偽造publish_timenow()寫入發布時間后續同步進 ES 的date字段。 FastAPI + Tortoise-ORM + Elasticsearch 實戰：職位數據全量同步到 ES 的設計
項目實踐FastAPI Elasticsearch 職位數據全量同步FastAPI Tortoise-ORM Elasticsearch 實戰職位數據全量同步到 ES 的設計意識與踩坑復盤一、前言介紹1.1 功能定位1.2 數據模型總覽1.3 同步流程總覽二、環境準備2.1 依賴與 ES 客戶端2.2 索引與運行環境2.3 路由與生命周期掛載三、知識點講解3.1 ORM 與 ES 的數據形態差異寬表思想3.2 IntEnumField 與 JSONField 的存儲語義3.3 text / keyword / date 三類字段選型3.4 異步客戶端單例與依賴注入3.5 批量寫入與冪等_id async_bulk3.6 N1 查詢的批量化解法四、代碼邏輯拆解4.1 ES 客戶端單例與依賴橋接4.2 值序列化工具 _to_es_value4.3 寬表文檔拼裝 _build_job_document4.4 創建索引mapping 設計4.5 全量同步 insert-data-v24.6 職位寫入接口 saveJobFastAPI Tortoise-ORM Elasticsearch 實戰職位數據全量同步到 ES 的設計意識與踩坑復盤一、前言介紹1.1 功能定位本文只講兩件事的設計意識其一職位領域模型與寫入接口怎么落地其二如何把 MySQL 里的職位數據批量、可靠地同步進 ES。1.2 數據模型總覽三個實體之間的關系是同步邏輯的基礎Job職位表 t_job └── enterprise_id : IntField ──┐ 手動整型關聯不做外鍵級聯 ↓ Enterprise企業主表 ── 1:1 ── EnterpriseInfo企業工商信息 └── industry : ForeignKeyField → Industry行業 Industry行業字典/ City城市字典要點Job與企業之間用的是enterprise_id普通整型字段而非ForeignKeyField這意味著同步時要手動按 ID 取企業且不會因為企業被刪而級聯掉職位。1.3 同步流程總覽啟動期lifespan 里初始化 ES 異步客戶端單例 ↓ 建索引POST /es-data/create-index-v2 → 聲明 mapping字段類型 IK 分詞 ↓ 全量同步POST /es-data/insert-data-v2 Job.all() → 收集 enterprise_id批量取 Enterprise / EnterpriseInfo 進內存字典 → 每條職位拼成寬表文檔 → async_bulk 批量寫入_idjob.id 保證冪等二、環境準備2.1 依賴與 ES 客戶端同步能力依賴官方elasticsearch異步客戶端。連接信息從環境變量讀取缺省回落到本機ES_HOSTos.getenv(ES_HOST,http://localhost:9200)es_client:AsyncElasticsearch|NoneNoneos.getenv(ES_HOST, ...)優先取環境變量便于不同環境切換 ES 地址es_client用模塊級None占位后續做單例整個進程只建一條連接。2.2 索引與運行環境ES 必須安裝IK 分詞插件ik_max_word否則analyzer: ik_max_word建索引會報錯本地開發可用單分片零副本number_of_shards: 1, number_of_replicas: 0職位索引取名boss_job_index_v2刻意與舊版boss_job_index分開便于對照學習。2.3 路由與生命周期掛載ES 客戶端在應用啟動期就初始化避免首次請求時再去建連asynccontextmanagerasyncdeflifespan(app:FastAPI):awaitTortoise.init(configTORTOISE_ORM,_enable_global_fallbackTrue)...awaitget_es_client()# 啟動即建立 ES 連接單例yieldawaitTortoise.close_connections()lifespan是 FastAPI 的啟動/關閉鉤子在yield之前做的都是啟動準備之后是優雅關閉這里只關了 TortoiseES 客戶端關閉函數雖已備好但并未在此顯式調用見問題排查 5.6。三、知識點講解3.1 ORM 與 ES 的數據形態差異寬表思想MySQL 里職位、企業、工商信息、行業分表存儲靠關聯還原。ES 是文檔型存儲更適合把一次搜索要展示的所有字段拍平成一條文檔避免搜索時再回查多表。這就是同步腳本把四張表拼成一份文檔的根本動機。3.2 IntEnumField 與 JSONField 的存儲語義statusfields.IntEnumField(enum_typeJobStatus,description0:草稿,1:招聘中...)department_idfields.IntEnumField(enum_typeDeptType,description所屬部門)job_tagsfields.JSONField(defaultlist,description職位標簽示例[五險一金,年終獎])IntEnumField數據庫存的是整數但 ORM 層自動轉成枚舉對象寫代碼用JobStatus.RECRUITING比裸數字更安全JSONField數據庫列里直接存 JSON 數組job_tags變成[五險一金,年終獎]進 ES 時映射成keyword多值字段。3.3 text / keyword / date 三類字段選型這是 mapping 設計的核心判斷text ik_max_word職位名、公司名、行業、經營范圍——需要中文全文檢索keyword薪資、城市、標簽、統一社會信用代碼——用于精確匹配 / 聚合 / 篩選不分詞integer / long狀態、部門、企業 ID——用于數值范圍與等值篩選date發布時間、認證時間——用于時間區間查詢寫入時用 ISO 字符串。3.4 異步客戶端單例與依賴注入asyncdefget_es_client()-AsyncElasticsearch:globales_clientifes_clientisNone:es_clientAsyncElasticsearch(ES_HOST)logger.info(ES 客戶端初始化成功)returnes_clientglobal es_client允許函數內修改模塊級變量二次進入直接返回已建好的實例進程內復用一條連接避免每次請求都握手接口層通過Depends(es_client_depend)拿到它與鑒權依賴寫法一致。3.5 批量寫入與冪等_id async_bulkactions.append({_index:BOSS_JOB_INDEX_NAME_V2,_id:str(job.id),_source:document,})awaitasync_bulk(clientes_client,actionsactions,raise_on_errorFalse)_idstr(job.id)把職位主鍵設為文檔 ID重復同步時覆蓋舊文檔而非新增天然冪等async_bulk一次性批量提交遠比逐條index()快raise_on_errorFalse單條失敗不中斷整批返回值里能拿到錯誤列表做統計。3.6 N1 查詢的批量化解法職位有 N 條若每條都現場查企業、查工商信息就是2N次查詢N1 的變體。正確做法是先收集所有enterprise_id兩批查完建字典內存里 O(1) 關聯enterprise_idslist({job.enterprise_idforjobinjobsifjob.enterprise_idisnotNone})enterprisesawaitEnterprise.filter(id__inenterprise_ids).prefetch_related(city)enterprise_map{e.id:eforeinenterprises}集合去重避免重復查詢prefetch_related(city)一次性把城市關聯出來后面取enterprise.city.name不再發 SQLenterprise_map以企業 ID 為鍵的字典拼文檔時直接enterprise_map.get(job.enterprise_id)。四、代碼邏輯拆解4.1 ES 客戶端單例與依賴橋接asyncdefes_client_depend()-AsyncElasticsearch:returnawaitget_es_client()這是 FastAPI 依賴函數作用是在路由簽名里優雅注入 ES 客戶端內部直接復用上面講的單例保證全鏈路同一連接。4.2 值序列化工具 _to_es_valueES 只認 JSON 友好的類型ORM 里的datetime、date、枚舉要先轉換def_to_es_value(value:Any)-Any:ifvalueisNone:returnNoneifisinstance(value,datetime):returnvalue.isoformat()ifisinstance(value,date):returnvalue.isoformat()ifisinstance(value,Enum):returnvalue.valuereturnvalue第 1 行空值原樣返回Nonemapping 里對應字段可空第 2–3 行datetime/date轉 ISO 字符串ESdate字段才能識別第 4 行Enum轉成.value一般是 int否則 ES 寫入枚舉對象會序列化失敗最后一行其余類型str、int、list、dict、None原樣透傳。4.3 寬表文檔拼裝 _build_job_document這是同步的靈魂——把職位、企業、工商、行業壓成一條扁平文檔citygetattr(enterprise,city,None)ifenterpriseelseNoneindustrygetattr(enterprise_info,industry,None)ifenterprise_infoelseNonereturn{job_id:job.id,job_name:job.job_name,min_salary:job.min_salary,max_salary:job.max_salary,job_tags:job.job_tagsor[],status:_to_es_value(job.status),publish_time:_to_es_value(job.publish_time),enterprise_name:enterprise.enterprise_nameifenterpriseelseNone,enterprise_account_status:_to_es_value(enterprise.account_status)ifenterpriseelseNone,enterpriseInfo_company_scale:_to_es_value(enterprise_info.company_scale)ifenterprise_infoelseNone,industry_id:industry.idifindustryelseNone,industry_name:industry.nameifindustryelseNone,}getattr(enterprise, city, None)防御式取值企業對象沒有city屬性時不拋異常job_tags or []標簽為空時給空數組避免 ES 收到None與keyword多值類型沖突if enterprise else None企業缺失時整組企業字段填None保證單條臟數據不會打斷整批字段名帶enterpriseInfo_前綴是為了和 mapping 一一對應也能直觀區分屬于企業詳情凡是status、publish_time、company_scale等枚舉/時間字段一律過_to_es_value。4.4 創建索引mapping 設計建索引時聲明每個字段類型與中文分詞器這些是經舊版踩坑后補全的job_name:{type:text,analyzer:ik_max_word},enterprise_name:{type:text,analyzer:ik_max_word},work_location:{type:keyword},min_salary:{type:keyword},status:{type:integer},publish_time:{type:date},enterpriseInfo_business_scope:{type:text,analyzer:ik_max_word},職位名、公司名、經營范圍用text ik_max_word支持中文全文檢索城市、薪資用keyword因為要做精確篩選與聚合不能分詞status用integer和IntEnumField存的整數對齊enterpriseInfo_business_scope舊版 mapping 漏聲明卻仍在寫入導致該字段無法被檢索這里顯式補回text類型建索引前先indices.exists判斷已存在則直接返回避免重復創建報錯。4.5 全量同步 insert-data-v2核心流程先校驗索引存在再批量取數再組裝 bulk最后一次性寫入。ifnotawaites_client.indices.exists(indexBOSS_JOB_INDEX_NAME_V2):return{code:0,message:f索引{BOSS_JOB_INDEX_NAME_V2}不存在請先調用 /es-data/create-index-v2}jobsawaitJob.all()ifnotjobs:return{code:1,message:沒有可同步的職位,data:{success:0,skip:0}}第一步先確認目標索引已建否則 bulk 時才報錯浪費前面查庫的開銷Job.all()一次取出全部職位空集合提前返回避免后面空跑。forjobinjobs:enterpriseenterprise_map.get(job.enterprise_id)enterprise_infoenterprise_info_map.get(job.enterprise_id)ifenterpriseisNone:skip_count1logger.warning(f同步跳過職位 id{job.id}關聯企業 id{job.enterprise_id}不存在)continueifenterprise_infoisNone:logger.warning(f職位 id{job.id}無企業詳情將寫入空的 enterpriseInfo_* 字段)document_build_job_document(job,enterprise,enterprise_info)actions.append({_index:BOSS_JOB_INDEX_NAME_V2,_id:str(job.id),_source:document})enterprise_map.get(...)內存字典取值O(1)無額外 SQL企業主數據缺失則continue跳過寬表缺核心信息寫入也搜不到意義不大并記日志企業詳情缺失允許繼續只是enterpriseInfo_*字段為None每條文檔帶_idstr(job.id)這是冪等覆蓋的關鍵全部 action 先攢進列表最后統一async_bulk。success_count,errorsawaitasync_bulk(clientes_client,actionsactions,raise_on_errorFalse,)error_countlen(errors)ifisinstance(errors,list)else0return{code:1,message:數據同步完成,data:{index:BOSS_JOB_INDEX_NAME_V2,job_total:len(jobs),success:success_count,skip:skip_count,error:error_count},}success_count是成功條數errors是失敗詳情列表raise_on_errorFalse讓部分失敗不影響整體返回里把成功/跳過/失敗三類數量都帶上便于對賬。4.6 職位寫入接口 saveJob職位落庫時企業 ID 不是前端傳的而是從登錄的招聘團隊反查出來避免越權掛到別家企業名下staticmethodasyncdefsaveJob(job:JobCreateRequest,time_id:int):timeawaitRecruitTeam.filter(idtime_id).first()awaitJob.create(**job.dict(),enterprise_idtime.enterprise_id,recruit_team_idtime_id,publish_timenow(),)time_id來自 JWT 鑒權依賴get_job_info代表當前招聘團隊RecruitTeam.filter(idtime_id).first()查出團隊取其enterprise_id**job.dict()把請求體字段展開成建表參數省去逐字段賦值enterprise_idtime.enterprise_id企業歸屬由服務端決定前端無法偽造publish_timenow()寫入發布時間后續同步進 ES 的date字段。