登入、获取用户list、添加用户、修改用户状态、删除用户

src/app/api/login.py

@@ -1,10 +1,9 @@
 from flask import request, jsonify
 from sqlalchemy import select
-from werkzeug.security import check_password_hash
 from flask_restx import Resource, Namespace, reqparse
 from sqlalchemy.orm import Session
 
-from app.configs.config import engine
+from app.database import engine
 from app.modle.users import User
 from app.utils.jwt_util import generate_jwt
 from app.defines import StatesCode

src/app/api/users.py

@@ -1,7 +1,11 @@
-from flask import request
+from flask import request, jsonify
 from flask_restx import Resource, Namespace, reqparse
+from sqlalchemy import insert, select, update, delete
+from sqlalchemy.orm import Session
 
 from app.defines import StatesCode
+from app.database import engine
+from app.modle.users import User
 
 ns = Namespace('users', description='用户管理接口')
 
@@ -20,46 +24,139 @@ class GetUserListApi(Resource):
         status = request.args.get('status')
         page_size = int(request.args.get('page_size', 20))
         page = int(request.args.get('page', 1))
-        data = [
-            {"user_name": "admin", "mobile": "13912345678", "email": "admin@qq.com", "role": "admin", "permission": 1,
-             "unit": "北京电信规划院", "registration_time": "2020-01-01 10:10"},
-            {"user_name": "admin1", "mobile": "13912345679", "email": "admin1@qq.com", "role": "admin", "permission": 1,
-             "unit": "北京电信", "registration_time": "2020-01-01 11:10"},
-        ]
+
+        if status is None:
+            return jsonify(code=StatesCode.UNKNOWN_ERROR, message='用户状态不能为空')
+
+        with Session(engine) as session:
+            stmt = select(User).where(User.account_status == status).offset(page_size * (page - 1)).limit(page_size)
+            results = session.execute(stmt).scalars().all()
+
+        data = []
+        for result in results:
+            result_dict = result.__dict__
+            del result_dict['_sa_instance_state']
+            result_dict['register_time'] = result_dict.get('register_time').strftime("%Y-%m-%d %H:%M:%S")
+            data.append(result_dict)
+
         return {"code": StatesCode.SUCCESS, "message": "成功", "data": data}
 
 
+add_user = reqparse.RequestParser(bundle_errors=True)
+add_user.add_argument(name='username', type=str, required=True, location='form', help='用户名')
+add_user.add_argument(name='password', type=str, required=True, location='form', help='密码')
+add_user.add_argument(name='photograph', type=str, required=False, location='form', help='照片')
+add_user.add_argument(name='name', type=str, required=False, location='form', help='姓名')
+add_user.add_argument(name='phone', type=str, required=False, location='form', help='电话')
+add_user.add_argument(name='email', type=str, required=False, location='form', help='邮箱')
+add_user.add_argument(name='company', type=str, required=False, location='form', help='公司')
+add_user.add_argument(name='department', type=str, required=False, location='form', help='部门')
+add_user.add_argument(name='role', type=str, required=False, location='form', help='角色')
+add_user.add_argument(name='permission', type=str, required=False, location='form', help='权限')
+add_user.add_argument(name='on_job_status', type=str, required=False, location='form', help='在职状态')
+add_user.add_argument(name='duty', type=str, required=False, location='form', help='职务')
+add_user.add_argument(name='nationality', type=str, required=False, location='form', help='国籍')
+
+users_status = reqparse.RequestParser(bundle_errors=True)
+users_status.add_argument(name='id', type=int, required=True, location='form', help='用户id')
+users_status.add_argument(name='status', type=int, required=True, location='form', help='用户状态')
+
+del_users = reqparse.RequestParser(bundle_errors=True)
+del_users.add_argument(name='id', type=int, required=True, location='form', help='用户id')
+
+
 @ns.route('/user')
 class UsersApi(Resource):
     @ns.doc(id='query_users', description='搜索用户')
     @ns.expect()
     def get(self):
         """搜索用户"""
-        data = [
-            {"user_name": "admin", "mobile": "13912345678", "email": "admin@qq.com", "role": "admin", "permission": 1,
-             "unit": "北京电信规划院", "registration_time": "2020-01-01 10:10"},
-            # {"user_name": "admin1", "mobile": "13912345679", "email": "admin1@qq.com", "role": "admin", "permission": 1,
-            #  "unit": "北京电信", "registration_time": "2020-01-01 11:10"},
-        ]
+        data = []
         return {"code": StatesCode.SUCCESS, "message": "成功", "data": data}
 
     @ns.doc(id='add_users', description='添加用户')
-    @ns.expect()
+    @ns.expect(add_user)
     def post(self):
         """添加用户"""
-        return {"code": StatesCode.SUCCESS, "message": "成功", "data": "user002"}
-
-    @ns.doc(id='modify_users_details', description='修改用户状态，激活、禁用')
-    @ns.expect()
+        username = request.form.get('username')
+        password = request.form.get('password')
+        photograph = request.form.get('photograph')
+        name = request.form.get('name')
+        phone = request.form.get('phone')
+        email = request.form.get('email')
+        company = request.form.get('company')
+        department = request.form.get('department')
+        role = request.form.get('role')
+        permission = request.form.get('permission')
+        on_job_status = request.form.get('on_job_status')
+        duty = request.form.get('duty')
+        nationality = request.form.get('nationality')
+
+        if username is None or password is None:
+            return jsonify(code=StatesCode.UNKNOWN_ERROR, message="用户名或密码不能为空")
+
+        with Session(engine) as session:
+
+            # 判断用户是否存在
+            stmt = select(User).where(User.user_name == username)
+            result = session.execute(stmt).scalars().first()
+            if result:
+                return jsonify(code=StatesCode.UNKNOWN_ERROR, message="用户已存在")
+
+            # 添加用户
+            session.execute(
+                insert(User).values(
+                    user_name=username,
+                    password=User().generate_password(password),
+                    photograph=photograph,
+                    name=name,
+                    phone=phone,
+                    email=email,
+                    company=company,
+                    department=department,
+                    role=role,
+                    permission=permission,
+                    on_job_status=on_job_status,
+                    duty=duty,
+                    nationality=nationality,
+
+                )
+            )
+
+            session.commit()
+
+        return jsonify(code=StatesCode.SUCCESS, message="添加用户成功")
+
+    @ns.doc(id='modify_users_status', description='修改用户状态，激活、禁用')
+    @ns.expect(users_status)
     def put(self):
         """修改用户状态，激活、禁用"""
-        return {"code": StatesCode.SUCCESS, "message": "成功", "data": ""}
+        user_id = int(request.form.get('id'))
+        account_status = int(request.form.get('status'))
+
+        if user_id is None or account_status is None:
+            return jsonify(code=StatesCode.UNKNOWN_ERROR, message="用户id或用户状态不能为空")
+
+        with Session(engine) as session:
+            session.execute(update(User).where(User.id == user_id).values(account_status=account_status))
+            session.commit()
+
+        return jsonify(code=StatesCode.SUCCESS, message="修改用户状态成功")
 
     @ns.doc(id='delete_users', description='删除用户')
-    @ns.expect()
+    @ns.expect(del_users)
     def delete(self):
         """删除用户"""
-        return {"code": StatesCode.SUCCESS, "message": "成功", "data": "user003"}
+        user_id = request.form.get('id')
+
+        if user_id is None:
+            return jsonify(code=StatesCode.UNKNOWN_ERROR, message="用户id不能为空")
+
+        with Session(engine) as session:
+            session.execute(delete(User).where(User.id == user_id))
+            session.commit()
+
+        return jsonify(code=StatesCode.SUCCESS, message="删除成功")
 
 
 @ns.route('/user_details')

src/app/database.py

@@ -0,0 +1,10 @@
+from sqlalchemy import create_engine
+
+user = 'mac'
+password = ''
+host = 'localhost'
+port = 5432
+database = 'postgres'
+uri = f'postgresql+psycopg2://{user}:{password}@{host}:{port}/{database}'
+
+engine = create_engine(uri)

src/app/modle/users.py

@@ -1,23 +1,22 @@
-from datetime import time
-
-from sqlalchemy import String, Column, Integer, DateTime
+from datetime import datetime
+import time
+from sqlalchemy import create_engine, String, Column, Integer, DateTime
+from werkzeug.security import check_password_hash, generate_password_hash
 
 from app.modle import Base
 
 
 def cnNow():
     """获取时间，年月日时分秒格式"""
-    # tz = pytz.timezone('Asia/Shanghai')
-    # return tz.fromutc(datetime.datetime.now() + datetime.timedelta(0, time.altzone))
-    return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
+    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
 
 
 class User(Base):
     """用户表"""
     __tablename__ = "users"
 
-    id = Column(Integer, primary_key=True, autoincrement=True, nullable=False, doc='用户id')
-    user_name = Column(String, nullable=False, unique=False, index=False, doc='用户名')
+    id = Column(Integer, primary_key=True, autoincrement=True, nullable=False, unique=True, doc='用户id')
+    user_name = Column(String, nullable=False, unique=True, index=False, doc='用户名')
     password = Column(String, nullable=False, unique=False, index=False, doc='密码')
     photograph = Column(String, nullable=True, unique=False, index=False, doc='照片')
     name = Column(String, nullable=True, unique=False, index=False, doc='姓名')
@@ -29,6 +28,15 @@ class User(Base):
     permission = Column(String, nullable=True, unique=False, index=False, doc='权限')
     on_job_status = Column(String, nullable=True, unique=False, index=False, doc='在职状态')
     duty = Column(String, nullable=True, unique=False, index=False, doc='职务')
-    account_status = Column(String, nullable=True, unique=False, index=False, doc='账户状态')
+    account_status = Column(Integer, nullable=True, unique=False, index=False, default=0,
+                            doc='账户状态,0为正常，1为禁用')
     nationality = Column(String, nullable=True, unique=False, index=False, doc='国籍')
     register_time = Column(DateTime, nullable=True, unique=False, index=False, default=cnNow(), doc='注册时间')
+
+    # 赋值password，则自动加密存储。
+    def generate_password(self, value):
+        return generate_password_hash(value)
+
+    # 使用check_password,进行密码校验，返回True False。
+    def check_password(self, pasword):
+        return check_password_hash(self.password, pasword)