优化pydantic v2兼容并测试

Author: amisadmin

README.md

@@ -282,10 +282,10 @@ flowchart LR
 from datetime import date
 
 from fastapi_amis_admin.models.fields import Field
-from fastapi_user_auth.auth.models import User
+from fastapi_user_auth.auth.models import BaseUser
 
-# 自定义`User`模型,继承`User`
-class MyUser(User, table = True):
+# 自定义`User`模型,继承`BaseUser`
+class MyUser(BaseUser, table = True):
     point: float = Field(default = 0, title = '积分', description = '用户积分')
     phone: str = Field(None, title = '手机号', max_length = 15)
     parent_id: int = Field(None, title = "上级", foreign_key = "auth_user.id")

fastapi_user_auth/admin/actions.py

@@ -124,7 +124,7 @@ class UpdateSubRolesAction(BaseSubAction):
 
     class schema(BaseModel):
         role_keys: str = Field(
-            None,
+            "",
             title="角色列表",
             amis_form_item=amis.Transfer(
                 selectMode="table",
@@ -198,7 +198,7 @@ class BaseSubPermAction(BaseSubAction):
     # 创建动作表单数据模型
     class schema(BaseModel):
         permissions: str = Field(
-            None,
+            "",
             title="权限列表",
             amis_form_item=amis.InputTree(
                 multiple=True,
@@ -279,7 +279,7 @@ class UpdateSubDataPermAction(BaseSubPermAction):
     # 创建动作表单数据模型
     class schema(BaseSubPermAction.schema):
         effect_matrix: list = Field(
-            None,
+            [],
             title="当前权限",
             amis_form_item=amis.MatrixCheckboxes(
                 rowLabel="权限名称",
@@ -290,7 +290,7 @@ class schema(BaseSubPermAction.schema):
             ),
         )
         policy_matrix: list = Field(
-            None,
+            [],
             title="权限配置",
             amis_form_item=amis.MatrixCheckboxes(
                 rowLabel="名称",

fastapi_user_auth/admin/admin.py

@@ -27,6 +27,7 @@
 from fastapi_amis_admin.amis.constants import DisplayModeEnum, LevelEnum
 from fastapi_amis_admin.crud.base import SchemaUpdateT
 from fastapi_amis_admin.crud.schema import BaseApiOut
+from fastapi_amis_admin.utils.pydantic import model_fields
 from fastapi_amis_admin.utils.translation import i18n as _
 from pydantic import BaseModel
 from sqlalchemy import select
@@ -238,8 +239,8 @@ async def get_form(self, request: Request) -> Form:
         form = await super().get_form(request)
         formitems = [
             await self.get_form_item(request, modelfield)
-            for k, modelfield in self.user_model.__fields__.items()
-            if k not in self.schema.__fields__.keys() | {"delete_time"}
+            for k, modelfield in model_fields(self.user_model).items()
+            if k not in model_fields(self.schema).keys() | {"delete_time"}
         ]
         form.body.extend(formitem.update_from_kwargs(disabled=True) for formitem in formitems if formitem)
         return form

fastapi_user_auth/auth/models.py

@@ -4,7 +4,7 @@
 from fastapi_amis_admin.crud.parser import LabelField
 from fastapi_amis_admin.models import Field
 from fastapi_amis_admin.utils.translation import i18n as _
-from sqlalchemy import ForeignKey, func, select
+from sqlalchemy import func, select
 
 from fastapi_user_auth.mixins.models import (  # noqa F401
     CreateTimeMixin,
@@ -49,16 +49,20 @@ class User(BaseUser, table=True):
     pass
 
 
-class Role(PkMixin, CUDTimeMixin, table=True):
-    """角色"""
-
+class BaseRole(PkMixin, CUDTimeMixin):
     __tablename__ = "auth_role"
 
     key: str = Field(title="角色标识", max_length=40, unique=True, index=True, nullable=False)
     name: str = Field(default="", title="角色名称", max_length=40)
     desc: str = Field(default="", title="角色描述", max_length=400, amis_form_item="textarea")
 
 
+class Role(BaseRole, table=True):
+    """角色"""
+
+    pass
+
+
 class CasbinRule(PkMixin, table=True):
     __tablename__ = "auth_casbin_rule"
 
@@ -115,7 +119,7 @@ class LoginHistory(PkMixin, CreateTimeMixin, table=True):
 
     __tablename__ = "auth_login_history"
 
-    user_id: int = Field(None, title="用户ID", sa_column_args=(ForeignKey("auth_user.id", ondelete="CASCADE"),))
+    user_id: Optional[int] = Field(None, title="用户ID")
     login_name: str = Field("", title="登录名", max_length=20)
     ip: str = Field("", title="登录IP", max_length=20)
     ip_info: str = Field("", title="IP信息", max_length=255)

fastapi_user_auth/auth/schemas.py

@@ -1,7 +1,9 @@
 from enum import Enum
+from typing import Optional
 
+from fastapi_amis_admin.utils.pydantic import PYDANTIC_V2
 from fastapi_amis_admin.utils.translation import i18n as _
-from pydantic import BaseModel, SecretStr, validator
+from pydantic import BaseModel, SecretStr
 from sqlmodel import Field
 
 from .models import BaseUser, EmailMixin, PasswordMixin, UsernameMixin
@@ -16,20 +18,32 @@ class UserLoginOut(BaseUser):
     """用户登录返回信息"""
 
     token_type: str = "bearer"
-    access_token: str = None
-    password: SecretStr = None
+    access_token: Optional[str] = None
+    password: Optional[SecretStr] = None
 
 
 class UserRegIn(UsernameMixin, PasswordMixin, EmailMixin):
     """用户注册"""
 
     password2: str = Field(title=_("Confirm Password"), max_length=128)
 
-    @validator("password2")
-    def passwords_match(cls, v, values, **kwargs):
-        if "password" in values and v != values["password"]:
-            raise ValueError("passwords do not match!")
-        return v
+    if PYDANTIC_V2:
+        from pydantic import model_validator
+
+        @model_validator(mode="after")
+        def check_passwords_match(self):
+            if self.password is not None and self.password.get_secret_value() != self.password2:
+                raise ValueError("passwords do not match!")
+            return self
+
+    else:
+        from pydantic import validator
+
+        @validator("password2")
+        def passwords_match_(cls, v, values, **kwargs):
+            if "password" in values and v != values["password"]:
+                raise ValueError("passwords do not match!")
+            return v
 
 
 # 默认保留的用户

fastapi_user_auth/mixins/models.py

@@ -1,11 +1,14 @@
 from datetime import datetime
 from typing import Optional
 
-from fastapi_amis_admin.models import Field, SQLModel
+from fastapi_amis_admin.models.fields import Field
 from fastapi_amis_admin.utils.translation import i18n as _
 from pydantic import EmailStr, SecretStr
 from sqlalchemy import func
 from sqlmodel import AutoString
+from sqlmodelx import SQLModel
+
+from fastapi_user_auth.utils.sqltypes import SecretStrType
 
 
 class PkMixin(SQLModel):
@@ -17,7 +20,7 @@ class CreateTimeMixin(SQLModel):
 
 
 class UpdateTimeMixin(SQLModel):
-    update_time: datetime = Field(
+    update_time: Optional[datetime] = Field(
         default_factory=datetime.now,
         title=_("Update Time"),
         sa_column_kwargs={"onupdate": func.now(), "server_default": func.now()},
@@ -38,13 +41,9 @@ class UsernameMixin(SQLModel):
     username: str = Field(title=_("Username"), max_length=32, unique=True, index=True, nullable=False)
 
 
-# class PasswordStr(SecretStr, str):
-#     pass
-
-
 class PasswordMixin(SQLModel):
     password: SecretStr = Field(
-        title=_("Password"), sa_type=AutoString, max_length=128, nullable=False, amis_form_item="input-password"
+        title=_("Password"), max_length=128, sa_type=SecretStrType, nullable=False, amis_form_item="input-password"
     )
 
 

fastapi_user_auth/utils/sqltypes.py

@@ -0,0 +1,31 @@
+from typing import cast
+
+from pydantic import SecretStr
+from sqlalchemy import String, types
+from sqlalchemy.engine import Dialect
+
+
+class SecretStrType(types.TypeDecorator):
+    impl = String
+    cache_ok = True
+    mysql_default_length = 255
+
+    @property
+    def python_type(self):
+        return self.impl.python_type
+
+    def load_dialect_impl(self, dialect: Dialect):
+        impl = cast(types.String, self.impl)
+        if impl.length is None and dialect.name == "mysql":
+            return dialect.type_descriptor(types.String(self.mysql_default_length))
+        return super().load_dialect_impl(dialect)
+
+    def process_bind_param(self, value, dialect):
+        if value and isinstance(value, SecretStr):
+            return value.get_secret_value()
+        return value
+
+    def process_result_value(self, value, dialect):
+        if value is not None:
+            return SecretStr(value)
+        return value

pyproject.toml

@@ -36,8 +36,7 @@ classifiers = [
     "Programming Language :: Python :: 3.11",
 ]
 dependencies = [
-    "pydantic>=1.10.0,<2.0.0",
-    "fastapi-amis-admin[sqlmodel]>=0.7.0,<0.8.0",
+    "fastapi-amis-admin>=0.6.0,<0.8.0",
     "email-validator>=1.3.1,<3.0.0",
     "passlib>=1.7.4",
     "bcrypt>=4.0.0",
@@ -51,15 +50,15 @@ FastAPI-Amis-Admin = "https://github.com/amisadmin/fastapi_amis_admin"
 
 [project.optional-dependencies]
 jwt = [
-    "python-jose==3.3.0",
+    "python-jose>=3.3.0",
 ]
 redis = ["redis>=4.2.0"]
 test = [
     "uvicorn[standard] >=0.19.0,<1.0",
     "pytest >=6.2.4",
     "pytest-asyncio>=0.17",
     "aiosqlite>=0.15.0",
-    "python-jose==3.3.0",
+    "python-jose>=3.3.0",
     "jinja2 >=2.11.2,<4.0.0",
     "ujson>=5.5.0",
     "requests>=2.28.1",

tests/test_auth/test_auth_user_model.py

@@ -17,9 +17,6 @@ class MyUser(BaseUser, table=True):
     location: str = Field("", title="位置")
 
 
-print(123)
-
-
 @pytest.mark.parametrize("logins", ["admin"], indirect=True)
 async def test_custom_user_model(fake_auth, logins):
     # 使用自定义的`User`模型,创建auth对象