refactor(backend): switch to pydantic from marshmallow

backend/app/config.py

@@ -5,7 +5,6 @@
 from sqlalchemy.engine import URL
 from sqlalchemy.event import listen
 from apispec import APISpec
-from apispec.ext.marshmallow import MarshmallowPlugin
 from apispec_webframeworks.flask import FlaskPlugin
 from prometheus_client import multiprocess
 from prometheus_client.core import CollectorRegistry
@@ -210,7 +209,7 @@ def get_secret(env_var: str, default: str = None) -> str | None:
         "description": "Find more info at the official documentation",
         "url": "https://docs.kitchenowl.org",
     },
-    plugins=[FlaskPlugin(), MarshmallowPlugin()],
+    plugins=[FlaskPlugin()],
 )
 api_spec.components.security_scheme(
     "bearerAuth", {"type": "http", "scheme": "bearer", "bearerFormat": "JWT"}

backend/app/controller/auth/auth_controller.py

@@ -59,7 +59,7 @@ def user_lookup_callback(_jwt_header, jwt_data) -> User | None:
 
     @auth.route("", methods=["POST"])
     @validate_args(Login)
-    def login(args):
+    def login(args: Login):
         """Authenticate user with username/email and password.
         ---
         post:
@@ -89,22 +89,22 @@ def login(args):
               description: Validation error
           security: []
         """
-        username = args["username"].lower().replace(" ", "")
+        username = args.username.lower().replace(" ", "")
         user = None
         if "@" not in username:
             user = User.find_by_username(username)
         else:
             user = User.find_by_email(username)
 
-        if not user or not user.check_password(args["password"]):
+        if not user or not user.check_password(args.password):
             raise UnauthorizedRequest(
                 message="Unauthorized: IP {} login attempt with wrong username or password".format(
                     getClientIp()
                 )
             )
         device = "Unkown"
         if "device" in args:
-            device = args["device"]
+            device = args.device
 
         # Create refresh token
         refreshToken, refreshModel = Token.create_refresh_token(user, device)
@@ -125,7 +125,7 @@ def login(args):
 
     @auth.route("signup", methods=["POST"])
     @validate_args(Signup)
-    def signup(args):
+    def signup(args: Signup):
         """Register new user account.
         ---
         post:
@@ -153,22 +153,19 @@ def signup(args):
               description: Invalid username/email
           security: []
         """
-        username = args["username"].lower().replace(" ", "").replace("@", "")
+        username = args.username.lower().replace(" ", "").replace("@", "")
         user = User.find_by_username(username)
         if user:
             return "Request invalid: username", 400
-        if "email" in args:
-            user = User.find_by_email(args["email"])
+        if args.email:
+            user = User.find_by_email(args.email)
             if user:
                 return "Request invalid: email", 400
 
         user = User.create(
-            username=username,
-            name=args["name"],
-            password=args["password"],
-            email=args["email"] if "email" in args else None,
+            username=username, name=args.name, password=args.password, email=args.email
         )
-        if "email" in args and mail.mailConfigured():
+        if args.email and mail.mailConfigured():
             gevent.spawn(
                 mail.sendVerificationMail,
                 user.id,
@@ -177,7 +174,7 @@ def signup(args):
 
         device = "Unkown"
         if "device" in args:
-            device = args["device"]
+            device = args.device
 
         # Create refresh token
         refreshToken, refreshModel = Token.create_refresh_token(user, device)
@@ -297,7 +294,7 @@ def logout(id):
 @auth.route("llt", methods=["POST"])
 @jwt_required()
 @validate_args(CreateLongLivedToken)
-def createLongLivedToken(args):
+def createLongLivedToken(args: CreateLongLivedToken):
     """Generate long-lived authentication token.
     ---
     post:
@@ -326,7 +323,7 @@ def createLongLivedToken(args):
     if not user:
         raise UnauthorizedRequest(message="Unauthorized: IP {}".format(getClientIp()))
 
-    llToken, _ = Token.create_longlived_token(user, args["device"])
+    llToken, _ = Token.create_longlived_token(user, args.device)
 
     return jsonify({"longlived_token": llToken})
 
@@ -378,7 +375,7 @@ def deleteLongLivedToken(id):
     @auth.route("oidc", methods=["GET"])
     @jwt_required(optional=True)
     @validate_args(GetOIDCLoginUrl)
-    def getOIDCLoginUrl(args):
+    def getOIDCLoginUrl(args: GetOIDCLoginUrl):
         """Generate OIDC authentication URL.
         ---
         get:
@@ -416,7 +413,7 @@ def getOIDCLoginUrl(args):
             - bearerAuth: []
             - {}
         """
-        provider = args["provider"] if "provider" in args else "custom"
+        provider = args.provider if args.provider else "custom"
         if provider not in oidc_clients:
             raise NotFoundRequest()
         client = oidc_clients[provider]
@@ -430,7 +427,7 @@ def getOIDCLoginUrl(args):
         nonce = rndstr()
         redirect_uri = (
             ("kitchenowl:" + ("" if OIDC_RFC_COMPLIANT_REDIRECT else "//"))
-            if "kitchenowl_scheme" in args and args["kitchenowl_scheme"]
+            if args.kitchenowl_scheme
             else FRONT_URL
         ) + "/signin/redirect"
         args = {
@@ -456,7 +453,7 @@ def getOIDCLoginUrl(args):
     @auth.route("callback", methods=["POST"])
     @jwt_required(optional=True)
     @validate_args(LoginOIDC)
-    def loginWithOIDC(args):
+    def loginWithOIDC(args: LoginOIDC):
         """Handle OIDC authentication callback.
         ---
         post:
@@ -492,7 +489,7 @@ def loginWithOIDC(args):
             - {}
         """
         # Validate oidc login
-        oidc_request = OIDCRequest.find_by_state(args["state"])
+        oidc_request = OIDCRequest.find_by_state(args.state)
         if not oidc_request:
             raise UnauthorizedRequest(
                 message="Unauthorized: IP {} login attempt with unknown OIDC state".format(
@@ -521,15 +518,15 @@ def loginWithOIDC(args):
 
         client.parse_response(
             AuthorizationResponse,
-            info={"code": args["code"], "state": oidc_request.state},
+            info={"code": args.code, "state": oidc_request.state},
             sformat="dict",
         )
 
         tokenResponse = client.do_access_token_request(
             scope=["openid", "profile", "email"],
             state=oidc_request.state,
             request_args={
-                "code": args["code"],
+                "code": args.code,
                 "redirect_uri": oidc_request.redirect_uri,
             },
             authn_method="client_secret_post",
@@ -622,8 +619,8 @@ def loginWithOIDC(args):
 
         # login user
         device = "Unkown"
-        if "device" in args:
-            device = args["device"]
+        if args.device:
+            device = args.device
 
         # Create refresh token
         refreshToken, refreshModel = Token.create_refresh_token(user, device)

backend/app/controller/auth/schemas.py

@@ -1,77 +1,42 @@
-from marshmallow import fields, Schema
+from typing import Annotated
+from pydantic import AfterValidator, BaseModel, EmailStr, StringConstraints
 
 from app.config import EMAIL_MANDATORY
+from app.helpers.validators import validate_non_emty_no_at
 
 
-class Login(Schema):
-    username = fields.String(required=True, validate=lambda a: a and not a.isspace())
-    password = fields.String(
-        required=True,
-        validate=lambda a: a and not a.isspace(),
-        load_only=True,
-    )
-    device = fields.String(
-        required=False,
-        validate=lambda a: a and not a.isspace(),
-        load_only=True,
-    )
+class Login(BaseModel):
+    username: Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)]
+    password: Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)]
+    device: Annotated[
+        str | None, StringConstraints(min_length=1, strip_whitespace=True)
+    ] = None
 
 
-class Signup(Schema):
-    username = fields.String(
-        required=True, validate=lambda a: a and not a.isspace() and "@" not in a
-    )
-    email = fields.String(
-        required=EMAIL_MANDATORY,
-        validate=lambda a: a and not a.isspace() and "@" in a,
-        load_only=True,
-    )
-    name = fields.String(required=True, validate=lambda a: a and not a.isspace())
-    password = fields.String(
-        required=True,
-        validate=lambda a: a and not a.isspace(),
-        load_only=True,
-    )
-    device = fields.String(
-        required=False,
-        validate=lambda a: a and not a.isspace(),
-        load_only=True,
-    )
+class Signup(BaseModel):
+    username: Annotated[str, AfterValidator(validate_non_emty_no_at)]
+    email: Annotated[
+        EmailStr | None, AfterValidator(lambda x: EMAIL_MANDATORY or x is not None)
+    ] = None
+    name: Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)]
+    password: Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)]
+    device: Annotated[
+        str | None, StringConstraints(min_length=1, strip_whitespace=True)
+    ] = None
 
 
-class CreateLongLivedToken(Schema):
-    device = fields.String(
-        required=True,
-        validate=lambda a: a and not a.isspace(),
-        load_only=True,
-    )
+class CreateLongLivedToken(BaseModel):
+    device: Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)]
 
 
-class GetOIDCLoginUrl(Schema):
-    provider = fields.String(
-        validate=lambda a: a and not a.isspace(),
-        load_only=True,
-    )
-    kitchenowl_scheme = fields.Boolean(
-        required=False,
-        load_default=False,
-        load_only=True,
-    )
+class GetOIDCLoginUrl(BaseModel):
+    provider: Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)]
+    kitchenowl_scheme: bool = False
 
 
-class LoginOIDC(Schema):
-    state = fields.String(
-        validate=lambda a: a and not a.isspace(),
-        required=True,
-        load_only=True,
-    )
-    code = fields.String(
-        validate=lambda a: a and not a.isspace(),
-        required=True,
-        load_only=True,
-    )
-    device = fields.String(
-        validate=lambda a: a and not a.isspace(),
-        required=False,
-        load_only=True,
-    )
+class LoginOIDC(BaseModel):
+    state: Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)]
+    code: Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)]
+    device: Annotated[
+        str | None, StringConstraints(min_length=1, strip_whitespace=True)
+    ] = None

backend/app/controller/category/category_controller.py

@@ -30,9 +30,9 @@ def getCategory(id):
 @jwt_required()
 @authorize_household()
 @validate_args(AddCategory)
-def addCategory(args, household_id):
+def addCategory(args: AddCategory, household_id):
     category = Category()
-    category.name = args["name"]
+    category.name = args.name
     category.household_id = household_id
     category.save()
     return jsonify(category.obj_to_dict())
@@ -41,20 +41,20 @@ def addCategory(args, household_id):
 @category.route("/<int:id>", methods=["POST", "PATCH"])
 @jwt_required()
 @validate_args(UpdateCategory)
-def updateCategory(args, id):
+def updateCategory(args: UpdateCategory, id):
     category = Category.find_by_id(id)
     if not category:
         raise NotFoundRequest()
     category.checkAuthorized()
 
-    if "name" in args:
-        category.name = args["name"]
-    if "ordering" in args and category.ordering != args["ordering"]:
-        category.reorder(args["ordering"])
+    if args.name is not None:
+        category.name = args.name
+    if args.ordering is not None and category.ordering != args.ordering:
+        category.reorder(args.ordering)
     category.save()
 
-    if "merge_category_id" in args and args["merge_category_id"] != id:
-        mergeCategory = Category.find_by_id(args["merge_category_id"])
+    if args.merge_category_id is not None and args.merge_category_id != id:
+        mergeCategory = Category.find_by_id(args.merge_category_id)
         if mergeCategory:
             category.merge(mergeCategory)
 
@@ -77,9 +77,9 @@ def deleteCategoryById(id):
 @jwt_required()
 @authorize_household()
 @validate_args(DeleteCategory)
-def deleteCategoryByName(args, household_id):
-    if "name" in args:
-        category = Category.find_by_name(args["name"], household_id)
+def deleteCategoryByName(args: DeleteCategory, household_id):
+    if args.name is not None:
+        category = Category.find_by_name(args.name, household_id)
         if category:
             category.delete()
             return jsonify({"msg": "DONE"})

backend/app/controller/category/schemas.py

@@ -1,23 +1,20 @@
-from marshmallow import fields, Schema, EXCLUDE
+from typing import Annotated
+from pydantic import BaseModel, StringConstraints, PositiveInt
 
 
-class AddCategory(Schema):
-    class Meta:
-        unknown = EXCLUDE
+class AddCategory(BaseModel):
+    name: Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)]
 
-    name = fields.String(required=True, validate=lambda a: a and not a.isspace())
 
-
-class UpdateCategory(Schema):
-    name = fields.String(validate=lambda a: a and not a.isspace())
-    ordering = fields.Integer(validate=lambda i: i >= 0)
+class UpdateCategory(BaseModel):
+    name: Annotated[
+        str | None, StringConstraints(min_length=1, strip_whitespace=True)
+    ] = None
+    ordering: PositiveInt | None = None
 
     # if set this merges the specified category into this category thus combining them to one
-    merge_category_id = fields.Integer(
-        validate=lambda a: a > 0,
-        allow_none=True,
-    )
+    merge_category_id: PositiveInt | None = None
 
 
-class DeleteCategory(Schema):
-    name = fields.String(required=True, validate=lambda a: a and not a.isspace())
+class DeleteCategory(BaseModel):
+    name: Annotated[str, StringConstraints(min_length=1, strip_whitespace=True)]