Fix the swagger form login structure abnormality (#46)

Author: wu-clan

backend/app/api/jwt.py

@@ -18,7 +18,7 @@
 
 pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
 
-oauth2_schema = OAuth2PasswordBearer(tokenUrl=settings.TOKEN_URL)
+oauth2_schema = OAuth2PasswordBearer(tokenUrl=settings.TOKEN_URL_SWAGGER)
 
 
 def get_hash_password(password: str) -> str:

backend/app/api/service/user_service.py

@@ -9,13 +9,13 @@
 from backend.app.crud.crud_user import UserDao
 from backend.app.database.db_mysql import async_db_session
 from backend.app.models import User
-from backend.app.schemas.user import CreateUser, ResetPassword, UpdateUser, Avatar
+from backend.app.schemas.user import CreateUser, ResetPassword, UpdateUser, Avatar, Auth
 from backend.app.utils import re_verify
 
 
 class UserService:
     @staticmethod
-    async def login(form_data: OAuth2PasswordRequestForm):
+    async def swagger_login(form_data: OAuth2PasswordRequestForm):
         async with async_db_session() as db:
             current_user = await UserDao.get_user_by_username(db, form_data.username)
             if not current_user:
@@ -32,23 +32,23 @@ async def login(form_data: OAuth2PasswordRequestForm):
             access_token = jwt.create_access_token(user.id)
             return access_token, user
 
-    # @staticmethod
-    # async def login(obj: Auth):
-    #     async with async_db_session() as db:
-    #         current_user = await UserDao.get_user_by_username(db, obj.username)
-    #         if not current_user:
-    #             raise errors.NotFoundError(msg='用户名不存在')
-    #         elif not jwt.password_verify(obj.password, current_user.password):
-    #             raise errors.AuthorizationError(msg='密码错误')
-    #         elif not current_user.is_active:
-    #             raise errors.AuthorizationError(msg='该用户已被锁定，无法登录')
-    #         # 更新登陆时间
-    #         await UserDao.update_user_login_time(db, obj.username)
-    #         # 获取最新用户信息
-    #         user = await UserDao.get_user_by_id(db, current_user.id)
-    #         # 创建token
-    #         access_token = jwt.create_access_token(user.id)
-    #         return access_token, user
+    @staticmethod
+    async def login(obj: Auth):
+        async with async_db_session() as db:
+            current_user = await UserDao.get_user_by_username(db, obj.username)
+            if not current_user:
+                raise errors.NotFoundError(msg='用户名不存在')
+            elif not jwt.password_verify(obj.password, current_user.password):
+                raise errors.AuthorizationError(msg='密码错误')
+            elif not current_user.is_active:
+                raise errors.AuthorizationError(msg='该用户已被锁定，无法登录')
+            # 更新登陆时间
+            await UserDao.update_user_login_time(db, obj.username)
+            # 获取最新用户信息
+            user = await UserDao.get_user_by_id(db, current_user.id)
+            # 创建token
+            access_token = jwt.create_access_token(user.id)
+            return access_token, user
 
     @staticmethod
     async def register(obj: CreateUser):

backend/app/api/v1/auth/user.py

@@ -8,23 +8,22 @@
 from backend.app.common.pagination import Page
 from backend.app.common.response.response_schema import response_base
 from backend.app.schemas.token import Token
-from backend.app.schemas.user import CreateUser, GetUserInfo, ResetPassword, UpdateUser, Avatar
+from backend.app.schemas.user import CreateUser, GetUserInfo, ResetPassword, UpdateUser, Avatar, Auth
 
 router = APIRouter()
 
 
-@router.post('/login', summary='表单登录', description='form 格式登录，支持直接在 api 文档调试接口')
-async def user_login(form_data: OAuth2PasswordRequestForm = Depends()):
-    token, user = await UserService.login(form_data)
-    data = Token(access_token=token, user=user)
-    return response_base.response_200(data=data)
+@router.post('/swagger_login', summary='swagger 表单登录', description='form 格式登录，仅用于 swagger 文档调试接口')
+async def swagger_user_login(form_data: OAuth2PasswordRequestForm = Depends()) -> Token:
+    token, user = await UserService.swagger_login(form_data)
+    return Token(access_token=token, user=user)
 
 
-# @router.post('/login', summary='用户登录', description='json 格式登录, 仅支持在第三方api工具调试接口, 例如: postman')
-# async def user_login(obj: Auth):
-#     token, user = await UserService.login(obj)
-#     data = Token(access_token=token, user=user)
-#     return response_base.response_200(data=data)
+@router.post('/login', summary='用户登录', description='json 格式登录, 仅支持在第三方api工具调试接口, 例如: postman')
+async def user_login(obj: Auth):
+    token, user = await UserService.login(obj)
+    data = Token(access_token=token, user=user)
+    return response_base.response_200(data=data)
 
 
 @router.post('/register', summary='用户注册')

backend/app/api/v1/sys_config.py

@@ -42,7 +42,7 @@ async def get_sys_config() -> ResponseModel:
             'aps_misfire_grace_time': settings.APS_MISFIRE_GRACE_TIME,
             'token_algorithm': settings.TOKEN_ALGORITHM,
             'token_expire_minutes': settings.TOKEN_EXPIRE_MINUTES,
-            'token_url': settings.TOKEN_URL,
+            'token_url': settings.TOKEN_URL_SWAGGER,
             'log_file_name': settings.LOG_FILE_NAME,
             'middleware_cors': settings.MIDDLEWARE_CORS,
             'middleware_gzip': settings.MIDDLEWARE_GZIP,

backend/app/core/conf.py

@@ -71,7 +71,7 @@ def validator_api_url(cls, values):
     # Token
     TOKEN_ALGORITHM: str = 'HS256'  # 算法
     TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 1  # token 时效 60 * 24 * 1 = 1 天
-    TOKEN_URL: str = '/v1/auth/users/login'
+    TOKEN_URL_SWAGGER: str = '/v1/auth/users/swagger_login'
 
     # Log
     LOG_FILE_NAME: str = 'fba.log'

backend/app/test/conftest.py

@@ -19,9 +19,9 @@ def anyio_backend():
 @pytest.fixture(scope='package', autouse=True)
 async def function_fixture(anyio_backend):
     auth_data = {
-        'url': f'http://{settings.UVICORN_HOST}:{settings.UVICORN_PORT}{settings.TOKEN_URL}',
-        'headers': {'accept': 'application/json', 'Content-Type': 'application/x-www-form-urlencoded'},
-        'data': {'username': 'test', 'password': 'test'},
+        'url': f'http://{settings.UVICORN_HOST}:{settings.UVICORN_PORT}/v1/auth/users/login',
+        'headers': {'accept': 'application/json', 'Content-Type': 'application/json'},
+        'json': {'username': 'test', 'password': 'test'},
     }
     async with AsyncClient() as client:
         response = await client.post(**auth_data)

backend/app/test/test_auth.py

@@ -25,10 +25,10 @@ async def get_token(self):
 
     async def test_login(self):
         async with AsyncClient(
-            app=app, headers={'accept': 'application/json', 'Content-Type': 'application/x-www-form-urlencoded'}
+            app=app, headers={'accept': 'application/json', 'Content-Type': 'application/json'}
         ) as client:
             response = await client.post(
-                url=f'{self.users_api_base_url}/login', data={'username': '1', 'password': '1'}
+                url=f'{self.users_api_base_url}/login', json={'username': 'test', 'password': 'test'}
             )
         assert response.status_code == 200
         assert response.json()['data']['token_type'] == 'Bearer'