perf: User api

Author: fit2cloud-chenyw

.env.example

@@ -13,6 +13,7 @@ [email protected]
 FIRST_SUPERUSER_PASSWORD=123456  # Change this to your pwd
 
 TOKEN_KEY="X-SQLBOT-TOKEN"
+DEFAULT_PWD="SQLBot@123456"
 
 # Postgres
 POSTGRES_SERVER=localhost

backend/apps/system/api/user.py

@@ -1,10 +1,11 @@
-from fastapi import APIRouter, Depends, Request    
-from apps.system.crud.user import get_user_info
-from apps.system.models.user import user_grid
-from apps.system.schemas.system_schema import UserLanguage
+from fastapi import APIRouter
+from apps.system.crud.user import get_db_user, get_user_info
+from apps.system.models.user import UserModel
+from apps.system.schemas.system_schema import UserCreator, UserEditor, UserGrid, UserLanguage
 from common.core.deps import CurrentUser, SessionDep
 from common.core.pagination import Paginator
 from common.core.schemas import PaginatedResponse, PaginationParams
+from common.utils.time import get_timestamp
 
 router = APIRouter(tags=["user"], prefix="/user")
 
@@ -18,7 +19,7 @@ async def user_info(session: SessionDep, current_user: CurrentUser):
     return db_user
 
 
-@router.get("/pager/{pageNum}/{pageSize}", response_model=PaginatedResponse[user_grid])
+@router.get("/pager/{pageNum}/{pageSize}", response_model=PaginatedResponse[UserGrid])
 async def pager(
     session: SessionDep,
     pageNum: int,
@@ -28,16 +29,44 @@ async def pager(
     paginator = Paginator(session)
     filters = {}
     return await paginator.get_paginated_response(
-        model=user_grid,
+        model=UserModel,
         pagination=pagination,
         **filters)
+
+@router.get("/{id}", response_model=UserEditor)
+async def query(session: SessionDep, id: int) -> UserEditor:
+    db_user: UserModel = get_db_user(session = session, user_id = id)
+    return db_user
+
+@router.post("")
+async def create(session: SessionDep, creator: UserCreator):
+    data = creator.model_dump(exclude_unset=True)
+    user_model = UserModel.model_validate(data)
+    #user_model.create_time = get_timestamp()
+    user_model.language = "zh-CN"
+    session.add(user_model)
+    session.commit()
+    
+@router.put("")
+async def update(session: SessionDep, editor: UserEditor):
+    user_model: UserModel = get_db_user(session = session, user_id = editor.id)
+    data = editor.model_dump(exclude_unset=True)
+    user_model.sqlmodel_update(data)
+    session.add(user_model)
+    session.commit()
+    
+@router.delete("/{id}") 
+async def delete(session: SessionDep, id: int):
+    user_model: UserModel = get_db_user(session = session, user_id = id)
+    session.delete(user_model)
+    session.commit()
 
 @router.put("/language")
 async def langChange(session: SessionDep, current_user: CurrentUser, language: UserLanguage):
     lang = language.language
     if lang not in ["zh-CN", "en"]:
         return {"message": "Language not supported"}
-    db_user = get_user_info(session=session, user_id=current_user.id)
+    db_user = session.get(UserModel, current_user.id)
     if not db_user:
         return {"message": "User not found"}
     db_user.language = lang

backend/apps/system/crud/user.py

@@ -1,22 +1,28 @@
 
 from sqlmodel import Session, select
-from ..models.user import sys_user, user_grid
+
+from apps.system.schemas.system_schema import BaseUserDTO
+from ..models.user import UserModel
 from common.core.security import verify_md5pwd
 
-def get_user_by_account(*, session: Session, account: str) -> sys_user | None:
-    #statement = select(sys_user).where(sys_user.account == account)
-    statement = select(user_grid.id, user_grid.account, user_grid.oid, user_grid.password).where(user_grid.account == account)
-    session_user = session.exec(statement).first()
-    result_user = sys_user.model_validate(session_user)
-    return result_user
+def get_db_user(*, session: Session, user_id: int) -> UserModel:
+    db_user = session.get(UserModel, user_id)
+    if not db_user:
+        raise RuntimeError("user not exist")
+    return db_user
 
-def get_user_info(*, session: Session, user_id: int) -> sys_user | None:
-    db_user = session.get(user_grid, user_id)
+def get_user_by_account(*, session: Session, account: str) -> BaseUserDTO | None:
+    statement = select(UserModel).where(UserModel.account == account)
+    db_user = session.exec(statement).first()
     if not db_user:
         return None
-    return db_user
+    return BaseUserDTO.model_validate(db_user.model_dump())
+
+def get_user_info(*, session: Session, user_id: int) -> BaseUserDTO | None:
+    db_user = get_db_user(session = session, user_id = user_id)
+    return BaseUserDTO.model_validate(db_user.model_dump())
 
-def authenticate(*, session: Session, account: str, password: str) -> sys_user | None:
+def authenticate(*, session: Session, account: str, password: str) -> BaseUserDTO | None:
     db_user = get_user_by_account(session=session, account=account)
     if not db_user:
         return None

backend/apps/system/models/user.py

@@ -1,25 +1,23 @@
 
 from typing import Optional
-from sqlmodel import SQLModel, Field
+from sqlmodel import BigInteger, SQLModel, Field
+
+from common.core.models import SnowflakeBase
+from common.core.security import default_md5_pwd
+from common.utils.time import get_timestamp
 
-class sys_user(SQLModel):
-    id: int = Field(primary_key=True, index=True)
-    account: str = Field(max_length=255, unique=True)
-    password: str = Field(max_length=255)
-    oid: int = Field(default=1)
 
-    def to_dict(self):
-        return {
-            "id": self.id,
-            "account": self.account,
-            "oid": self.oid
-        }
-        
-class user_grid(sys_user, table=True):
-    __tablename__ = "sys_user"
-    name: str
-    email: str
-    status: int
-    create_time: int
-    language: str = Field(max_length=255, default="zh-CN")
 
+class BaseUserPO(SQLModel):
+    account: str = Field(max_length=255, unique=True)
+    oid: int = Field(nullable=False, sa_type=BigInteger())
+    name: str = Field(max_length=255, unique=True)
+    password: str = Field(default_factory=default_md5_pwd, max_length=255)
+    email: str = Field(max_length=255)
+    status: int = Field(default=0, nullable=False)
+    create_time: int = Field(default_factory=get_timestamp, sa_type=BigInteger(), nullable=False)
+    language: str = Field(max_length=255, default="zh-CN")
+    
+class UserModel(SnowflakeBase, BaseUserPO, table=True):
+    __tablename__ = "sys_user"
+    

backend/apps/system/schemas/system_schema.py

@@ -1,9 +1,39 @@
 from pydantic import BaseModel
 
+from common.core.schemas import BaseCreatorDTO
+
 class model_status(BaseModel):
     status: bool
     ids: list[int]
 
 
 class UserLanguage(BaseModel):
-    language: str
+    language: str
+    
+
+class BaseUser(BaseModel):
+    account: str
+    oid: int
+
+class BaseUserDTO(BaseUser, BaseCreatorDTO):
+    password: str
+    def to_dict(self):
+        return {
+            "id": self.id,
+            "account": self.account,
+            "oid": self.oid
+        }
+
+
+
+class UserCreator(BaseUser):
+    name: str
+    email: str
+    status: int = 1
+
+class UserEditor(UserCreator, BaseCreatorDTO):
+   pass
+
+class UserGrid(UserEditor):
+    create_time: int
+    language: str = "zh-CN"

backend/common/core/config.py

@@ -58,6 +58,7 @@ def all_cors_origins(self) -> list[str]:
     POSTGRES_DB: str = ""
 
     TOKEN_KEY: str
+    DEFAULT_PWD: str
 
     @computed_field  # type: ignore[prop-decorator]
     @property

backend/common/core/deps.py

@@ -4,13 +4,15 @@
 from fastapi import Depends, HTTPException, status
 # from fastapi.security import OAuth2PasswordBearer
 from jwt.exceptions import InvalidTokenError
-from pydantic import BaseModel, ValidationError
-from sqlmodel import Session, select
+from pydantic import ValidationError
+from sqlmodel import Session
+from apps.system.crud.user import get_db_user
+from apps.system.schemas.system_schema import BaseUserDTO
 from common.core.schemas import TokenPayload, XOAuth2PasswordBearer
 from common.core import security
 from common.core.config import settings
 from common.core.db import get_session
-from apps.system.models.user import sys_user, user_grid
+from apps.system.models.user import UserModel
 reusable_oauth2 = XOAuth2PasswordBearer(
     tokenUrl=f"{settings.API_V1_STR}/login/access-token"
 )
@@ -21,7 +23,7 @@
 SessionDep = Annotated[Session, Depends(get_session)]
 TokenDep = Annotated[str, Depends(reusable_oauth2)]
 
-async def get_current_user(session: SessionDep, token: TokenDep) -> sys_user:
+async def get_current_user(session: SessionDep, token: TokenDep) -> BaseUserDTO:
     try:
         payload = jwt.decode(
             token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
@@ -32,17 +34,16 @@ async def get_current_user(session: SessionDep, token: TokenDep) -> sys_user:
             status_code=status.HTTP_403_FORBIDDEN,
             detail="Could not validate credentials",
         )
-    statement = select(user_grid.id, user_grid.account, user_grid.oid, user_grid.password).where(user_grid.id == token_data.id)
-    session_user = session.exec(statement).first()
+    session_user: UserModel = get_db_user(session = session, user_id = token_data.id)
     if not session_user:
         raise HTTPException(status_code=404, detail="User not found")
-    user = sys_user.model_validate(session_user)
+    user = BaseUserDTO.model_validate(session_user.model_dump())
     if not user:
         raise HTTPException(status_code=404, detail="User not found")
     """ if not user.is_active:
         raise HTTPException(status_code=400, detail="Inactive user") """
     return user
-CurrentUser = Annotated[sys_user, Depends(get_current_user)]
+CurrentUser = Annotated[BaseUserDTO, Depends(get_current_user)]
 
 
 

backend/common/core/security.py

@@ -35,3 +35,10 @@ def md5pwd(password: str) -> str:
 
 def verify_md5pwd(plain_password: str, md5_password: str) -> bool:
     return md5pwd(plain_password) == md5_password
+
+def default_pwd() -> str:
+    return settings.DEFAULT_PWD
+
+def default_md5_pwd() -> str:
+    pwd = default_pwd()
+    return md5pwd(pwd)

backend/common/utils/time.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 
 
-def get_timestamp():
+def get_timestamp() -> int:
     dt_millis = int(datetime.now().timestamp() * 1000)
     return dt_millis

backend/common/utils/whitelist.py

@@ -26,6 +26,7 @@
     "/system/config/key",
     "/images/*",
     "/sse",
+    "/user*"
 ]
 
 class WhitelistChecker: