80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
import logging
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy.orm import Session
|
|
|
|
try:
|
|
from app.db import get_db
|
|
from app.security import create_access_token
|
|
from app.service.parent import ParentService
|
|
from app.service.wechat_login import (
|
|
WechatAuthError,
|
|
WechatAuthService,
|
|
get_wechat_auth_service,
|
|
)
|
|
except ModuleNotFoundError:
|
|
from db import get_db
|
|
from security import create_access_token
|
|
from service.parent import ParentService
|
|
from service.wechat_login import WechatAuthError, WechatAuthService, get_wechat_auth_service
|
|
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
logger = logging.getLogger("app.auth")
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
code: str = Field(min_length=1, max_length=191)
|
|
nickname: str | None = Field(default=None, max_length=64)
|
|
avatar_url: str | None = Field(default=None, max_length=255)
|
|
|
|
|
|
class LoginResponse(BaseModel):
|
|
access_token: str
|
|
token_type: str = "bearer"
|
|
expires_in: int
|
|
user_id: int
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
def login(
|
|
payload: LoginRequest,
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
wechat_auth_service: WechatAuthService = Depends(get_wechat_auth_service),
|
|
) -> LoginResponse:
|
|
del request
|
|
logger.info("/auth/login called")
|
|
|
|
try:
|
|
wechat_session = wechat_auth_service.exchange_code(payload.code)
|
|
except WechatAuthError as exc:
|
|
logger.warning(
|
|
"wechat login failed",
|
|
extra={
|
|
"event": "wechat_login_failed",
|
|
"status_code": exc.status_code,
|
|
"errcode": exc.errcode,
|
|
},
|
|
)
|
|
raise HTTPException(status_code=exc.status_code, detail=str(exc)) from exc
|
|
|
|
parent_service = ParentService(db)
|
|
parent = parent_service.create(
|
|
openid=wechat_session.openid,
|
|
unionid=wechat_session.unionid,
|
|
nickname=payload.nickname,
|
|
avatar_url=payload.avatar_url,
|
|
)
|
|
user_id = int(parent["user_id"])
|
|
|
|
access_token, expires_in = create_access_token(user_id=user_id)
|
|
logger.info("wechat login succeeded", extra={"event": "wechat_login_succeeded", "user_id": user_id})
|
|
return LoginResponse(access_token=access_token, expires_in=expires_in, user_id=user_id)
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout(request: Request):
|
|
return {"message": "logged out"}
|