feat: Implement core infrastructure, authentication, models, GitLab, Jenkins, SonarQube, and Zentao plugins, along with comprehensive documentation and tests.

Author: netxs2000

devops_collector/auth/router.py

@@ -1,4 +1,7 @@
-"""TODO: Add module description."""
+"""认证模块路由。
+
+处理用户注册、登录、获取当前用户信息以及 GitLab OAuth 绑定。
+"""
 from datetime import timedelta, datetime
 from typing import Optional
 from fastapi import APIRouter, Depends, HTTPException, status, Request
@@ -10,28 +13,19 @@
 from devops_collector.models.base_models import User, UserOAuthToken
 from devops_collector.auth.database import SessionLocal
 from devops_collector.config import Config
+
 router = APIRouter(prefix='/auth', tags=['Authentication'])
 
 def get_db():
-    '''"""TODO: Add description.
-
-Args:
-    TODO
-
-Returns:
-    TODO
-
-Raises:
-    TODO
-"""'''
+    """获取数据库会话的依赖项。"""
     db = SessionLocal()
     try:
         yield db
     finally:
         db.close()
 
 @router.get('/gitlab/bind')
-async def bind_gitlab(request: Request, token: str=Depends(services.oauth2_scheme), db: Session=Depends(get_db)):
+async def bind_gitlab(request: Request, token: str = Depends(services.oauth2_scheme), db: Session = Depends(get_db)):
     """发起 GitLab OAuth 绑定。"""
     if not Config.GITLAB_CLIENT_ID or not Config.GITLAB_REDIRECT_URI:
         raise HTTPException(500, 'GitLab OAuth not configured')
@@ -43,24 +37,46 @@ async def bind_gitlab(request: Request, token: str=Depends(services.oauth2_schem
             raise HTTPException(401, 'User not found')
     except Exception:
         raise HTTPException(401, 'Invalid token')
+    
     state = str(current_user.global_user_id)
-    auth_url = f'{Config.GITLAB_URL}/oauth/authorize?client_id={Config.GITLAB_CLIENT_ID}&redirect_uri={Config.GITLAB_REDIRECT_URI}&response_type=code&scope=api&state={state}'
+    auth_url = (
+        f'{Config.GITLAB_URL}/oauth/authorize?'
+        f'client_id={Config.GITLAB_CLIENT_ID}&'
+        f'redirect_uri={Config.GITLAB_REDIRECT_URI}&'
+        f'response_type=code&scope=api&state={state}'
+    )
     return RedirectResponse(auth_url)
 
 @router.get('/gitlab/callback')
-async def gitlab_callback(code: str, state: str=None, db: Session=Depends(get_db)):
+async def gitlab_callback(code: str, state: str = None, db: Session = Depends(get_db)):
     """GitLab OAuth 回调处理。"""
     async with httpx.AsyncClient() as client:
-        resp = await client.post(f'{Config.GITLAB_URL}/oauth/token', data={'client_id': Config.GITLAB_CLIENT_ID, 'client_secret': Config.GITLAB_CLIENT_SECRET, 'code': code, 'grant_type': 'authorization_code', 'redirect_uri': Config.GITLAB_REDIRECT_URI})
+        resp = await client.post(
+            f'{Config.GITLAB_URL}/oauth/token', 
+            data={
+                'client_id': Config.GITLAB_CLIENT_ID, 
+                'client_secret': Config.GITLAB_CLIENT_SECRET, 
+                'code': code, 
+                'grant_type': 'authorization_code', 
+                'redirect_uri': Config.GITLAB_REDIRECT_URI
+            }
+        )
         if resp.status_code != 200:
             raise HTTPException(400, f'GitLab Auth Failed: {resp.text}')
         token_data = resp.json()
+    
     user_id = state
     if not user_id:
         raise HTTPException(400, 'Invalid State')
+    
     token_rec = db.query(UserOAuthToken).filter_by(user_id=user_id, provider='gitlab').first()
     if not token_rec:
-        token_rec = UserOAuthToken(user_id=user_id, provider='gitlab', access_token=token_data['access_token'], token_type=token_data.get('token_type', 'Bearer'))
+        token_rec = UserOAuthToken(
+            user_id=user_id, 
+            provider='gitlab', 
+            access_token=token_data['access_token'], 
+            token_type=token_data.get('token_type', 'Bearer')
+        )
         db.add(token_rec)
     else:
         token_rec.access_token = token_data['access_token']
@@ -69,70 +85,46 @@ async def gitlab_callback(code: str, state: str=None, db: Session=Depends(get_db
     return RedirectResponse(url='/static/iteration.html?bind_success=true')
 
 @router.post('/register', response_model=schemas.UserResponse)
-def register(user: schemas.UserRegisterRequest, db: Session=Depends(get_db)):
-    '''"""TODO: Add description.
-
-Args:
-    user: TODO
-    db: TODO
-
-Returns:
-    TODO
-
-Raises:
-    TODO
-"""'''
+def register(user: schemas.UserRegisterRequest, db: Session = Depends(get_db)):
+    """注册新用户。"""
     db_user = services.get_user_by_email(db, email=user.email)
     if db_user:
         raise HTTPException(status_code=400, detail='Email already registered')
     return services.create_user(db=db, user_data=user)
 
 @router.post('/login', response_model=schemas.Token)
-def login_for_access_token(form_data: OAuth2PasswordRequestForm=Depends(), db: Session=Depends(get_db)):
-    '''"""TODO: Add description.
-
-Args:
-    form_data: TODO
-    db: TODO
-
-Returns:
-    TODO
-
-Raises:
-    TODO
-"""'''
+def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
+    """登录获取访问令牌。"""
     user = services.authenticate_user(db, form_data.username, form_data.password)
     if not user:
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Incorrect username or password', headers={'WWW-Authenticate': 'Bearer'})
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED, 
+            detail='Incorrect username or password', 
+            headers={'WWW-Authenticate': 'Bearer'}
+        )
     access_token_expires = timedelta(minutes=services.ACCESS_TOKEN_EXPIRE_MINUTES)
-    access_token = services.create_access_token(data={'sub': user.primary_email, 'user_id': str(user.global_user_id)}, expires_delta=access_token_expires)
+    access_token = services.create_access_token(
+        data={'sub': user.primary_email, 'user_id': str(user.global_user_id)}, 
+        expires_delta=access_token_expires
+    )
     return {'access_token': access_token, 'token_type': 'bearer'}
 
 @router.get('/me', response_model=schemas.UserResponse)
-def read_users_me(token: str=Depends(services.oauth2_scheme), db: Session=Depends(get_db)):
-    '''"""TODO: Add description.
-
-Args:
-    token: TODO
-    db: TODO
-
-Returns:
-    TODO
-
-Raises:
-    TODO
-"""'''
+def read_users_me(token: str = Depends(services.oauth2_scheme), db: Session = Depends(get_db)):
+    """获取当前登录用户信息。"""
     try:
         payload = services.jwt.decode(token, services.SECRET_KEY, algorithms=[services.ALGORITHM])
         email: str = payload.get('sub')
         if email is None:
             raise HTTPException(status_code=401, detail='Invalid token')
     except Exception:
         raise HTTPException(status_code=401, detail='Invalid token')
+    
     user = services.get_user_by_email(db, email=email)
     if user is None:
         raise HTTPException(status_code=401, detail='User not found')
-    token = db.query(UserOAuthToken).filter_by(user_id=user.global_user_id, provider='gitlab').first()
+    
+    token_obj = db.query(UserOAuthToken).filter_by(user_id=user.global_user_id, provider='gitlab').first()
     resp = schemas.UserResponse.from_orm(user)
-    resp.gitlab_connected = True if token else False
+    resp.gitlab_connected = True if token_obj else False
     return resp

devops_collector/auth/services.py

@@ -1,6 +1,7 @@
-"""TODO: Add module description."""
-from datetime import datetime, timedelta
-from typing import Optional
+import uuid
+import logging
+from datetime import datetime, timedelta, timezone
+from typing import Optional, Any, Union
 from jose import JWTError, jwt
 from passlib.context import CryptContext
 from sqlalchemy.orm import Session
@@ -9,95 +10,54 @@
 from devops_collector.models.base_models import User, UserCredential
 from devops_collector.config import Config
 from uuid import UUID
+
 SECRET_KEY = getattr(Config, 'SECRET_KEY', 'your-secret-key-keep-it-secret')
 ALGORITHM = 'HS256'
 ACCESS_TOKEN_EXPIRE_MINUTES = 30
 pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl='auth/login')
 
-def verify_password(plain_password, hashed_password):
-    '''"""TODO: Add description.
-
-Args:
-    plain_password: TODO
-    hashed_password: TODO
-
-Returns:
-    TODO
-
-Raises:
-    TODO
-"""'''
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """验证明文密码是否与哈希密码匹配。"""
     return pwd_context.verify(plain_password, hashed_password)
 
-def get_password_hash(password):
-    '''"""TODO: Add description.
-
-Args:
-    password: TODO
-
-Returns:
-    TODO
-
-Raises:
-    TODO
-"""'''
+def get_password_hash(password: str) -> str:
+    """生成密码的 BCRPYT 哈希。"""
     return pwd_context.hash(password)
 
-def create_access_token(data: dict, expires_delta: Optional[timedelta]=None):
-    '''"""TODO: Add description.
-
-Args:
-    data: TODO
-    expires_delta: TODO
-
-Returns:
-    TODO
-
-Raises:
-    TODO
-"""'''
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
+    """生成 JWT 访问令牌。"""
     to_encode = data.copy()
     if expires_delta:
-        expire = datetime.utcnow() + expires_delta
+        expire = datetime.now(timezone.utc) + expires_delta
     else:
-        expire = datetime.utcnow() + timedelta(minutes=15)
+        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
+    
+    to_encode.update({"exp": expire})
     for k, v in to_encode.items():
-        if isinstance(v, UUID):
+        if isinstance(v, (UUID, uuid.UUID)):
             to_encode[k] = str(v)
     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
     return encoded_jwt
 
-def get_user_by_email(db: Session, email: str):
-    '''"""TODO: Add description.
-
-Args:
-    db: TODO
-    email: TODO
-
-Returns:
-    TODO
-
-Raises:
-    TODO
-"""'''
+def get_user_by_email(db: Session, email: str) -> Optional[User]:
+    """根据主邮箱获取当前有效用户。"""
     return db.query(User).filter(User.primary_email == email, User.is_current == True).first()
 
-def create_user(db: Session, user_data):
-    '''"""TODO: Add description.
-
-Args:
-    db: TODO
-    user_data: TODO
-
-Returns:
-    TODO
-
-Raises:
-    TODO
-"""'''
+def create_user(db: Session, user_data: Any) -> User:
+    """创建新用户及其认证凭据。"""
     hashed_password = get_password_hash(user_data.password)
-    db_user = User(global_user_id=uuid.uuid4(), primary_email=user_data.email, full_name=user_data.full_name, employee_id=user_data.employee_id, is_active=True, is_survivor=True, sync_version=1, is_current=True, is_deleted=False)
+    db_user = User(
+        global_user_id=uuid.uuid4(), 
+        primary_email=user_data.email, 
+        full_name=user_data.full_name, 
+        employee_id=user_data.employee_id, 
+        is_active=True, 
+        is_survivor=True, 
+        sync_version=1, 
+        is_current=True, 
+        is_deleted=False
+    )
     db.add(db_user)
     db.flush()
     db_cred = UserCredential(user_id=db_user.global_user_id, password_hash=hashed_password)
@@ -106,20 +66,8 @@ def create_user(db: Session, user_data):
     db.refresh(db_user)
     return db_user
 
-def authenticate_user(db: Session, email: str, password: str):
-    '''"""TODO: Add description.
-
-Args:
-    db: TODO
-    email: TODO
-    password: TODO
-
-Returns:
-    TODO
-
-Raises:
-    TODO
-"""'''
+def authenticate_user(db: Session, email: str, password: str) -> Union[User, bool]:
+    """验证用户凭据并返回用户对象。"""
     user = get_user_by_email(db, email)
     if not user:
         return False
@@ -129,18 +77,8 @@ def authenticate_user(db: Session, email: str, password: str):
         return False
     return user
 
-def get_current_user(token: str=Depends(oauth2_scheme)):
-    '''"""TODO: Add description.
-
-Args:
-    token: TODO
-
-Returns:
-    TODO
-
-Raises:
-    TODO
-"""'''
+def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
+    """FastAPI 依赖项：从令牌中获取当前认证用户。"""
     from devops_collector.auth.database import SessionLocal
     auth_db = SessionLocal()
     try:
@@ -158,18 +96,8 @@ def get_current_user(token: str=Depends(oauth2_scheme)):
     finally:
         auth_db.close()
 
-def get_current_active_user(current_user: User=Depends(get_current_user)):
-    '''"""TODO: Add description.
-
-Args:
-    current_user: TODO
-
-Returns:
-    TODO
-
-Raises:
-    TODO
-"""'''
+def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
+    """FastAPI 依赖项：验证当前用户是否处于激活状态。"""
     if not current_user.is_active:
         raise HTTPException(status_code=400, detail='Inactive user')
     return current_user