超级管理员

Author: caitinggui

apps/forms.py

@@ -0,0 +1,43 @@
+# -*- coding: utf-8 -*-
+
+import re
+
+from .models import User
+from utils import ParamsError, Constant
+
+
+EMAIL_PATTERN = re.compile(
+    "^.+\\@(\\[?)[a-zA-Z0-9\\-\\.]+\\.([a-zA-Z]{2,3}|[0-9]{1,3})(\\]?)$")
+
+
+def checkEmail(email):
+    if not email:
+        raise(ParamsError("email error"))
+    if len(email) > User.email.max_length or not EMAIL_PATTERN.match(email):
+        raise(ParamsError('email error'))
+    return True
+
+
+def checkPassword(password):
+    if not password:
+        raise(ParamsError("password error"))
+    if len(password) < Constant.password_min_length or len(password) > Constant.password_max_length:
+        raise ParamsError("password length must between %s and %",
+                          Constant.password_min_length, Constant.password_max_length)
+    return True
+
+
+def checkUsername(name):
+    if not name:
+        raise(ParamsError("name error"))
+    if len(name) > User.name.max_length:
+        raise(ParamsError("name too long"))
+    return True
+
+
+def checkSex(sex):
+    if not sex:
+        return False
+    if sex not in User.sex.choices:
+        raise(ParamsError("sex error"))
+    return True

apps/models.py

@@ -62,6 +62,7 @@ class Permission(object):
     EDIT = 4
     DELETE = 5
     ADD = 6
+    COMMENT = 7
 
 
 class Role(BaseModel):
@@ -73,6 +74,15 @@ class Role(BaseModel):
     created_time = DateTimeField(default=datetime.datetime.now)
     updated_time = DateTimeField(default=datetime.datetime.now)
 
+    @staticmethod
+    def initAdmin():
+        role = Role.create(name="admin", description="super admin", permission=Permission.ADMIN)
+        return role
+
+    class Meta:
+        db_table = 'role'
+
+
 
 class User(BaseModel):
     """表名都用单数"""
@@ -224,7 +234,7 @@ def createAllTables():
     s_user_todos = f"select id, {title}, {detail}, {is_completed}, {created_time}, {updated_time}, {edit_num} from {todo_table} where {user_id}=%s and {is_deleted}=0"
     s_user_todo = f"select id, {title}, {detail}, {is_completed}, {created_time}, {updated_time}, {edit_num} from {todo_table} where {user_id}=%s and {is_deleted}=0 and id=%s"
     s_todo_title = f"select id, {title} from {todo_table} where title=%s and {user_id}=%s and {is_deleted}=0"
-    s_all_todo = f"select id, A.{title}, A.{detail}, A.{created_time}, A.{updated_time}, A.{edit_num}, B.{username} from {todo_table} A {user_table} B where A.{user_id}=B.id and {is_deleted}=0 group by B.{username}"
+    s_all_todo = f"select A.id, A.{title}, A.{detail}, A.{created_time}, A.{updated_time}, A.{edit_num}, B.{username} from {todo_table} A, {user_table} B where A.{user_id}=B.id and A.{is_deleted}=0 and B.{is_deleted}=0 order by B.id"
 
     i_user = f"insert into {user_table} ({username}, {password}, {email}, {age}, {sex}, {city}, {signature}, {created_time}, {updated_time}, {last_login}) values (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
     i_todo = f"insert into {todo_table} ({user_id}, {title}, {detail}, {created_time}, {updated_time}) values (%s, %s, %s, %s, %s)"

apps/todo/view.py

@@ -4,7 +4,7 @@
 
 from sanic import Blueprint
 from .. import app
-from ..models import S
+from ..models import S, User
 from ..auth import login_required
 from ..baseview import BaseView
 from .forms import postTodoListView

apps/user/forms.py

@@ -1,18 +1,15 @@
 # -*- coding: utf-8 -*-
 
 import logging
-import re
 
 from ..models import User
+from ..forms import checkEmail, checkPassword, checkUsername, checkSex
 
 from utils import AttrDict, ParamsError, safeInt, Constant
 
 
 logger = logging.getLogger(__name__)
 
-EMAIL_PATTERN = re.compile(
-    "^.+\\@(\\[?)[a-zA-Z0-9\\-\\.]+\\.([a-zA-Z]{2,3}|[0-9]{1,3})(\\]?)$")
-
 
 def getTokenView(request):
     """处理TokenView的get方法的表单"""
@@ -26,31 +23,18 @@ def postTokenView(request):
 def postUserListView(request):
     name = request.form.get("name")
     password = request.form.get("password")
-    if None in (name, password):
-        raise(ParamsError)
-    if len(name) > User.name.max_length:
-        raise(ParamsError("name too long"))
-    if len(password) < Constant.password_min_length or len(password) > Constant.password_max_length:
-        raise ParamsError("password length must between %s and %",
-                          Constant.password_min_length, Constant.password_max_length)
-    password = User.generalPassword(password)
-
     email = request.form.get("email")
-    if email:
-        if len(email) > User.email.max_length or not EMAIL_PATTERN.match(email):
-            raise(ParamsError("email error"))
-    else:
-        email = User.email.default
+    checkUsername(name)
+    checkPassword(password)
+    checkEmail(email)
+    password = User.generalPassword(password)
 
     age = safeInt(request.form.get("age", User.age.default))
     if age < Constant.age_range[0] or age > Constant.age_range[1]:
         raise(ParamsError("age error"))
     # 先不给默认值
     sex = request.form.get("sex")
-    if sex:
-        if sex not in User.sex.choices:
-            raise(ParamsError("sex error"))
-    else:
+    if not checkSex(sex):
         sex = User.sex.choices.unknown
     city = request.form.get("city", User.city.default)
     if len(city) > User.city.max_length:

apps/user/view.py

@@ -44,7 +44,6 @@ async def post(self, request, **kwargs):
                 sql_data = await cur.fetchone()
                 if not sql_data:
                     return webJson(RetCode.PARAMETER_ERROR, data="无此用户")
-                logger.debug("sql data: %s", sql_data)
                 if User.verifyPassword(password, sql_data.password):
                     data = {Constant.user_id: sql_data.id}
                     data = User.generalToken(data, expiration=Constant.expires_in_login)

stodo_server.py

@@ -51,8 +51,28 @@ def cli():
 
 @cli.command(help="初始化数据库")
 def initdb():
-    from apps.models import S
+    from apps.models import S, Role
     S.createAllTables()
+    Role.initAdmin()
+
+
+@cli.command("createSuperUser", help="创建超级管理员账号")
+def createSuperUser():
+    from apps.models import User, Role
+    from apps.forms import checkEmail, checkUsername, checkPassword
+    # 从数据库绑定admin id
+    role_admin = Role.select(Role.id).where(Role.name=="admin").get().id
+    if not role_admin:
+        return("必须先在Role表中创建管理员角色")
+    email = click.prompt("请输入邮箱")
+    checkEmail(email)
+    name = click.prompt("请输入用户名")
+    checkUsername(name)
+    password = click.prompt("请输入密码", hide_input=True, confirmation_prompt=True)
+    checkPassword(password)
+    password = User.generalPassword(password)
+    User.create(email=email, name=name, password=password, sex=User.sex.choices.unknown, role_id=role_admin, signature="I am super admin")
+    print("成功创建超级管理员账户；%s" % name)
 
 
 @cli.command(help="运行server")