add test

api/sys_monitor_api.py

@@ -20,7 +20,6 @@ async def websocket_endpoint(
  websocket: WebSocket
 ):
  await manager.connect(websocket)
- logger.info("ws开始连接....")
  logger.info("websocket client ip==>{} port===>{}".format(websocket.client.host, websocket.client.port))
  try:
  while True:
@@ -32,5 +31,4 @@ async def websocket_endpoint(
  except asyncio.TimeoutError:
  await manager.send_json(message, websocket)
  except (WebSocketDisconnect, RuntimeError):
- logger.info("ws开始关闭连接.....")
  manager.disconnect(websocket)

api/user_api.py

@@ -6,13 +6,13 @@
 
 from fastapi import APIRouter, Body, Depends, Query
 from typing_extensions import Annotated
-
+from utils.jwt_token import parse_token
 from config import settings
 from config.factory import AutomationResponse
-from entity.user_entity import UserFrom
+from entity.user_entity import UserFrom, ActiveUser
 from sqlalchemy.orm import Session
 from config.db_session import get_db
-from repository.user_repository import login_user
+from repository.user_repository import login_user, active_user
 from service.user_service import user_service
 from utils.jwt_token import create_token
 from utils.response import success_200, error_211
@@ -24,7 +24,7 @@ router = APIRouter(prefix="/api", tags=["用户模块"])
 
 @router.post("/register/user", summary="注册用户", name="注册用户")
 async def register(user: Annotated[UserFrom, Body(...)], db: Session = Depends(get_db)):
- user = user_service(db, user)
+ user = await user_service(db, user)
  if isinstance(user, dict):
  return success_200(data=user, message="用户注册成功")
  return error_211(message=user)
@@ -39,3 +39,17 @@ async def login(user: OAuth2PasswordRequestForm = Depends(), db: Session = Depen
  return {"code": 200, "message": "登录成功", "access_token": token,
  "user": AutomationResponse.model_to_dict(user[1], "password"), "token_type": "Bearer"}
  return error_211(message=user[1])
+
+
+@router.post("/user/active", summary="用户启用和禁用", name="用户启用和禁用")
+async def active(user: Annotated[ActiveUser, Body(...)], db: Session = Depends(get_db), user_id: int = Depends(parse_token)):
+ status = 0
+ if user.status == 0: # 正在启用
+ status = 1
+ if user.status == 1: # 正在禁用
+ status = 0
+ active_user(db, user_id=user.id, status=status)
+ if status == 0: # 正在启用
+ return success_200(data=dict(status=1), message="禁用成功")
+ if status == 1: # 正在停用
+ return success_200(data=dict(status=0), message="启用成功")

entity/user_entity.py

@@ -10,3 +10,10 @@ class UserFrom(BaseModel):
  username: str
  password: str
  email: EmailStr
+ status: int = 1
+
+
+class ActiveUser(BaseModel):
+ id: int
+ status: int
+

models/model.py

@@ -4,7 +4,7 @@
 # @Author :qiangyanwen
 # @File :model.py
 from datetime import datetime
-from sqlalchemy import Column, INT, String, DATETIME
+from sqlalchemy import Column, INT, String, DATETIME,Integer
 from config.database import DatabaseModel
 
 
@@ -17,6 +17,7 @@ class User(DatabaseModel):
  username = Column(String(16), unique=True, index=True, comment="用户名")
  password = Column(String(256), comment="密码")
  email = Column(String(64), unique=True, nullable=False, comment="邮箱")
+ status = Column(Integer,nullable=False,comment="用户状态,0是禁用1是启用",default=1)
  created_time = Column(DATETIME, comment='创建时间')
  deleted_time = Column(DATETIME, comment="更新时间")
  __table_args__ = ({'comment': '用户表'})

repository/user_repository.py

@@ -6,7 +6,7 @@
 from sqlalchemy import or_
 from sqlalchemy.orm import Session
 
-from config.db_session import db_add
+from config.db_session import db_add, db_save
 from config.factory import AutomationResponse
 from entity.user_entity import UserFrom
 from enums.enums import RegisterUser
@@ -33,3 +33,9 @@ def login_user(db: Session, username, password):
  if user.username and user.email:
  return True, user
  return False, RegisterUser.LOGIN_USER_ERROR.value
+
+
+def active_user(db: Session, user_id: int, status: int):
+ user = db.query(User).filter(User.id == user_id).first()
+ user.status = status
+ db_save(db, user)

service/user_service.py

@@ -11,7 +11,7 @@ from utils.jwt_token import get_md5_pwd
 from repository.user_repository import register_user
 
 
-def user_service(db: Session, user: UserFrom):
+async def user_service(db: Session, user: UserFrom):
  user.password = get_md5_pwd(user.password)
  is_true = check_user_email(db, user)
  if is_true: