From f38df083db303d71d6aaf4d2db9259d104f27186 Mon Sep 17 00:00:00 2001 From: Xiaoxia AI Date: Fri, 19 Jun 2026 09:47:20 +0800 Subject: [PATCH] fix(phase7): resolve remaining TODOs - session_id in JWT and repository injection --- apps/api/app/api/routes/auth.py | 19 +++- .../worker/worker_app/tasks/classification.py | 89 ++++++++++--------- packages/application/auth/login_use_case.py | 17 ++-- 3 files changed, 74 insertions(+), 51 deletions(-) diff --git a/apps/api/app/api/routes/auth.py b/apps/api/app/api/routes/auth.py index 69f71d672..ea8b0a672 100644 --- a/apps/api/app/api/routes/auth.py +++ b/apps/api/app/api/routes/auth.py @@ -1,7 +1,7 @@ """ 认证 API 路由 """ -from fastapi import APIRouter, HTTPException, status, Depends +from fastapi import APIRouter, HTTPException, status, Depends, Request from pydantic import BaseModel, EmailStr from packages.application.auth import ( @@ -145,6 +145,7 @@ async def login(request: LoginRequestModel): @router.post("/logout", status_code=status.HTTP_204_NO_CONTENT) async def logout( + request: Request, logout_all_devices: bool = False, current_user: User = Depends(get_current_user), ): @@ -157,9 +158,23 @@ async def logout( container = get_container() use_case = container.get_logout_use_case() + # 从 JWT token 中提取 session_id + from packages.domain.auth import jwt_service + + # 从 request 中获取 token + auth_header = request.headers.get("Authorization") + session_id = None + if auth_header and auth_header.startswith("Bearer "): + token = auth_header[7:] # 移除 "Bearer " 前缀 + try: + payload = jwt_service.verify_token(token) + session_id = payload.get("sid") # 从 payload 提取 session_id + except: + pass # token 无效或没有 session_id，继续使用 None + req = LogoutRequest( user_id=current_user.id, - session_id=None, # TODO: 从 token 中获取 session_id + session_id=session_id, logout_all_devices=logout_all_devices, ) diff --git a/apps/worker/worker_app/tasks/classification.py b/apps/worker/worker_app/tasks/classification.py index 650f4dac5..e9c15cae8 100644 --- a/apps/worker/worker_app/tasks/classification.py +++ b/apps/worker/worker_app/tasks/classification.py @@ -1,6 +1,7 @@ from worker_app.celery_app import celery_app +from worker_app.db import SessionLocal from packages.domain import AssetClassification, ClassificationJob, ClassificationJobStatus -from packages.adapters.in_memory import InMemoryClassificationJobRepository +from packages.adapters.sqlalchemy_impl.classification_job_repository import SQLAlchemyClassificationJobRepository @celery_app.task(name="worker.classify_asset") @@ -15,45 +16,53 @@ def classify_asset(job_id: str) -> dict: 4. Update ClassificationJob with result 5. Return result """ - # TODO: Replace with real repository injection - job_repo = InMemoryClassificationJobRepository() - - job = job_repo.get(job_id) - if job is None: - return {"status": "failed", "error": "job not found"} - + # 创建数据库 session 和 repository + session = SessionLocal() try: - # Update job status to PROCESSING - job.status = ClassificationJobStatus.PROCESSING - job_repo.update(job) + job_repo = SQLAlchemyClassificationJobRepository(session) - # Mock classification (in real implementation: use ML model, vision API, etc.) - # For now, randomly classify based on asset_id hash - asset_id_hash = sum(ord(c) for c in job.asset_id) - classifications = list(AssetClassification) - classification = classifications[asset_id_hash % len(classifications)] - confidence = 0.85 + job = job_repo.get(job_id) + if job is None: + return {"status": "failed", "error": "job not found"} - # Update job status to COMPLETED - job.status = ClassificationJobStatus.COMPLETED - job.classification = classification.value - job.confidence = confidence - job_repo.update(job) - - return { - "status": "completed", - "job_id": job.id, - "classification": classification.value, - "confidence": confidence, - } - except Exception as e: - # Update job status to FAILED - job.status = ClassificationJobStatus.FAILED - job.error_message = str(e) - job_repo.update(job) - - return { - "status": "failed", - "job_id": job.id, - "error": str(e), - } + try: + # Update job status to PROCESSING + job.status = ClassificationJobStatus.PROCESSING + job_repo.update(job) + session.commit() + + # Mock classification (in real implementation: use ML model, vision API, etc.) + # For now, randomly classify based on asset_id hash + asset_id_hash = sum(ord(c) for c in job.asset_id) + classifications = list(AssetClassification) + classification = classifications[asset_id_hash % len(classifications)] + confidence = 0.85 + + # Update job status to COMPLETED + job.status = ClassificationJobStatus.COMPLETED + job.classification = classification.value + job.confidence = confidence + job_repo.update(job) + session.commit() + + return { + "status": "completed", + "job_id": job.id, + "classification": classification.value, + "confidence": confidence, + } + except Exception as e: + session.rollback() + # Update job status to FAILED + job.status = ClassificationJobStatus.FAILED + job.error_message = str(e) + job_repo.update(job) + session.commit() + + return { + "status": "failed", + "job_id": job.id, + "error": str(e), + } + finally: + session.close() diff --git a/packages/application/auth/login_use_case.py b/packages/application/auth/login_use_case.py index 3cdf600a6..188d7a8c8 100644 --- a/packages/application/auth/login_use_case.py +++ b/packages/application/auth/login_use_case.py @@ -87,13 +87,18 @@ class LoginUseCase: # if not user.email_verified: # return None, "Please verify your email first" - # 5. 生成基础 JWT token（不包含 workspace，登录后需要选择工作空间） + # 5. 创建 session 并生成 refresh_token + session_id = secrets.token_urlsafe(16) + refresh_token = secrets.token_urlsafe(32) + + # 6. 生成基础 JWT token（包含 session_id，不包含 workspace） # 这里使用一个特殊的 "user_token"，不包含 workspace 和 role # 用户选择工作空间后，会换取包含 workspace 的 access_token import jwt as pyjwt now = datetime.now(timezone.utc) access_token_payload = { "sub": user.id, + "sid": session_id, # 添加 session_id "type": "user_auth", # 标记为用户认证 token（未绑定工作空间） "iat": now, "exp": now + timedelta(minutes=jwt_service.config.ACCESS_TOKEN_EXPIRE_MINUTES), @@ -103,12 +108,6 @@ class LoginUseCase: jwt_service.config.SECRET_KEY, algorithm=jwt_service.config.ALGORITHM ) - - # 生成 refresh_token - refresh_token = secrets.token_urlsafe(32) - - # 6. 创建 session 并保存 refresh_token - session_id = secrets.token_urlsafe(16) session_store.save_session( session_id=session_id, user_id=user.id, @@ -118,12 +117,12 @@ class LoginUseCase: expires_in_seconds=30 * 24 * 3600, # 30 天 ) - # 7. 更新最后登录信息 + # 8. 更新最后登录信息 user.last_login_at = datetime.now(timezone.utc) user.last_login_ip = request.ip_address self.user_repository.save(user) - # 8. 返回响应 + # 9. 返回响应 return LoginResponse( access_token=access_token, refresh_token=refresh_token,