feat(routers): 完善auth、card、file、user相关接口开发

core/routers/auth/services.py

@@ -4,6 +4,66 @@
 # @Software : PyCharm
 # @Author : xingc
 # @Desc :
+from datetime import datetime, timedelta
+from typing import Annotated
+
+import jwt
+import ujson
+from fastapi import HTTPException, status, Depends
 from fastapi.security import OAuth2PasswordBearer
+import redis.asyncio as aioredis
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from config import settings
+from core.db.crud import CRUD
+from core.db.engine import get_async_session, get_cache
+from core.db.models import User
+from core.utils.helper import generate_keyname
+
+from . import schema
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+
+def create_access_token(data: dict):
+    to_encode = data.copy()
+    expire = datetime.utcnow() + timedelta(
+        minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
+    )
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, settings.ACCESS_SECRET_KEY, algorithm=settings.ALGORITHM)
+    return {'token_type': 'bearer', 'access_token': encoded_jwt}
+
+
+def verify_access_token(token: str):
+    try:
+        # payload = jwt.decode(token, settings.ACCESS_SECRET_KEY, algorithms=[settings.ALGORITHM])
+        payload = jwt.decode(token, settings.ACCESS_SECRET_KEY, algorithms=[settings.ALGORITHM])
+        return payload
+    except jwt.ExpiredSignatureError:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token has expired")
+    except jwt.InvalidTokenError:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
+
+
+async def get_current_user(
+        token: Annotated[str, Depends(oauth2_scheme)],
+        db_session: Annotated[AsyncSession, Depends(get_async_session)],
+        redis_conn: aioredis = Depends(get_cache)
+) -> schema.LoginUser:
+    """验证登录状态、获取的用户基础信息"""
+    result = verify_access_token(token)
+    user_id = result.get("sub")
+    if user_id is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
+
+    keyname = generate_keyname(f'login_cache', use_today=True)
+    user_str = await redis_conn.hget(keyname, user_id)
+    if user_str is None:
+        user_model = await CRUD(db_session, User).get(filters={'user_id': user_id})
+        user = schema.LoginUser(**user_model.dict())
+        await redis_conn.hset(keyname, user_id, user.json())
+    else:
+        user = schema.LoginUser(**ujson.decode(user_str))
+
+    return user