♻️ CODE REFACTOR (#40)

* 🙈 Updated Gitignore

Added .idea to gitignore file

* ♻️ REFACTOR CORS ORIGIN VALIDATION in Settings

refactored backend_cors_origin field in config to use annotated and BeforeValidator instead of the field validator decorator anymore and changed rstrip to strip in the

* ♻️ Switched from using Optional[Type] to Type | None

According to best practices and PEP 604, instead of Optional[Type], one should use Type | None, argument backed by [Link](https://discuss.python.org/t/clarification-for-pep-604-is-foo-int-none-to-replace-all-use-of-foo-optional-int/26945)

Author: thoth2357

.gitignore

@@ -4,4 +4,5 @@ testing-project
 # poetry.lock
 dev-link/
 
-.DS_Store
+.DS_Store
+.idea/

{{cookiecutter.project_slug}}/backend/app/app/core/config.py

@@ -1,10 +1,16 @@
 import secrets
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Union, Annotated
 
-from pydantic import AnyHttpUrl, EmailStr, HttpUrl, field_validator
+from pydantic import AnyHttpUrl, EmailStr, HttpUrl, field_validator, BeforeValidator
 from pydantic_core.core_schema import ValidationInfo
 from pydantic_settings import BaseSettings
 
+def parse_cors(v: Any) -> list[str] | str:
+    if isinstance(v, str) and not v.startswith("["):
+        return [i.strip() for i in v.split(",")]
+    elif isinstance(v, list | str):
+        return v
+    raise ValueError(v)
 
 class Settings(BaseSettings):
     API_V1_STR: str = "/api/v1"
@@ -21,21 +27,15 @@ class Settings(BaseSettings):
     # BACKEND_CORS_ORIGINS is a JSON-formatted list of origins
     # e.g: '["http://localhost", "http://localhost:4200", "http://localhost:3000", \
     # "http://localhost:8080", "http://local.dockertoolbox.tiangolo.com"]'
-    BACKEND_CORS_ORIGINS: List[AnyHttpUrl] = []
-
-    @field_validator("BACKEND_CORS_ORIGINS", mode="before")
-    def assemble_cors_origins(cls, v: Union[str, List[str]]) -> Union[List[str], str]:
-        if isinstance(v, str) and not v.startswith("["):
-            return [i.strip() for i in v.split(",")]
-        elif isinstance(v, (list, str)):
-            return v
-        raise ValueError(v)
+    BACKEND_CORS_ORIGINS: Annotated[
+        list[AnyHttpUrl] | str, BeforeValidator(parse_cors)
+    ] = []
 
     PROJECT_NAME: str
-    SENTRY_DSN: Optional[HttpUrl] = None
+    SENTRY_DSN: HttpUrl | None = None
 
     @field_validator("SENTRY_DSN", mode="before")
-    def sentry_dsn_can_be_blank(cls, v: str) -> Optional[str]:
+    def sentry_dsn_can_be_blank(cls, v: str) -> str | None:
         if isinstance(v, str) and len(v) == 0:
             return None
         return v
@@ -49,16 +49,16 @@ def sentry_dsn_can_be_blank(cls, v: str) -> Optional[str]:
     MONGO_DATABASE_URI: str
 
     SMTP_TLS: bool = True
-    SMTP_PORT: Optional[int] = None
-    SMTP_HOST: Optional[str] = None
-    SMTP_USER: Optional[str] = None
-    SMTP_PASSWORD: Optional[str] = None
-    EMAILS_FROM_EMAIL: Optional[EmailStr] = None
-    EMAILS_FROM_NAME: Optional[str] = None
-    EMAILS_TO_EMAIL: Optional[EmailStr] = None
+    SMTP_PORT: int = 587
+    SMTP_HOST: str | None = None
+    SMTP_USER: str | None = None
+    SMTP_PASSWORD: str | None = None
+    EMAILS_FROM_EMAIL: EmailStr | None = None
+    EMAILS_FROM_NAME: str | None = None
+    EMAILS_TO_EMAIL: EmailStr | None = None
 
     @field_validator("EMAILS_FROM_NAME")
-    def get_project_name(cls, v: Optional[str], info: ValidationInfo) -> str:
+    def get_project_name(cls, v: str | None, info: ValidationInfo) -> str:
         if not v:
             return info.data["PROJECT_NAME"]
         return v

{{cookiecutter.project_slug}}/backend/app/app/core/security.py

@@ -1,5 +1,5 @@
 from datetime import datetime, timedelta
-from typing import Any, Union, Optional
+from typing import Any, Union
 
 from jose import jwt
 from passlib.context import CryptContext
@@ -64,7 +64,7 @@ def create_magic_tokens(*, subject: Union[str, Any], expires_delta: timedelta =
     return magic_tokens
 
 
-def create_new_totp(*, label: str, uri: Optional[str] = None) -> NewTOTP:
+def create_new_totp(*, label: str, uri: str | None = None) -> NewTOTP:
     if not uri:
         totp = totp_factory.new()
     else:

{{cookiecutter.project_slug}}/backend/app/app/crud/base.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Generic, Optional, Type, TypeVar, Union
+from typing import Any, Dict, Generic, Type, TypeVar, Union
 
 from fastapi.encoders import jsonable_encoder
 from pydantic import BaseModel
@@ -27,20 +27,20 @@ def __init__(self, model: Type[ModelType]):
         self.model = model
         self.engine: AIOEngine = get_engine()
 
-    async def get(self, db: AgnosticDatabase, id: Any) -> Optional[ModelType]:
+    async def get(self, db: AgnosticDatabase, id: Any) -> ModelType | None:
         return await self.engine.find_one(self.model, self.model.id == id)
 
-    async def get_multi(self, db: AgnosticDatabase, *, page: int = 0, page_break: bool = False) -> list[ModelType]:
-        offset = {"skip": page * settings.MULTI_MAX, "limit": settings.MULTI_MAX} if page_break else {}
+    async def get_multi(self, db: AgnosticDatabase, *, page: int = 0, page_break: bool = False) -> list[ModelType]: # noqa
+        offset = {"skip": page * settings.MULTI_MAX, "limit": settings.MULTI_MAX} if page_break else {} # noqa
         return await self.engine.find(self.model, **offset)
 
-    async def create(self, db: AgnosticDatabase, *, obj_in: CreateSchemaType) -> ModelType:
+    async def create(self, db: AgnosticDatabase, *, obj_in: CreateSchemaType) -> ModelType: # noqa
         obj_in_data = jsonable_encoder(obj_in)
         db_obj = self.model(**obj_in_data)  # type: ignore
         return await self.engine.save(db_obj)
 
     async def update(
-        self, db: AgnosticDatabase, *, db_obj: ModelType, obj_in: Union[UpdateSchemaType, Dict[str, Any]]
+        self, db: AgnosticDatabase, *, db_obj: ModelType, obj_in: Union[UpdateSchemaType, Dict[str, Any]] # noqa
     ) -> ModelType:
         obj_data = jsonable_encoder(db_obj)
         if isinstance(obj_in, dict):

{{cookiecutter.project_slug}}/backend/app/app/crud/crud_user.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, Union
 
 from motor.core import AgnosticDatabase
 
@@ -11,22 +11,22 @@
 
 # ODM, Schema, Schema
 class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
-    async def get_by_email(self, db: AgnosticDatabase, *, email: str) -> Optional[User]:
+    async def get_by_email(self, db: AgnosticDatabase, *, email: str) -> User | None: # noqa
         return await self.engine.find_one(User, User.email == email)
 
-    async def create(self, db: AgnosticDatabase, *, obj_in: UserCreate) -> User:
+    async def create(self, db: AgnosticDatabase, *, obj_in: UserCreate) -> User: # noqa
         # TODO: Figure out what happens when you have a unique key like 'email'
         user = {
             **obj_in.model_dump(),
             "email": obj_in.email,
-            "hashed_password": get_password_hash(obj_in.password) if obj_in.password is not None else None,
+            "hashed_password": get_password_hash(obj_in.password) if obj_in.password is not None else None, # noqa
             "full_name": obj_in.full_name,
             "is_superuser": obj_in.is_superuser,
         }
 
         return await self.engine.save(User(**user))
 
-    async def update(self, db: AgnosticDatabase, *, db_obj: User, obj_in: Union[UserUpdate, Dict[str, Any]]) -> User:
+    async def update(self, db: AgnosticDatabase, *, db_obj: User, obj_in: Union[UserUpdate, Dict[str, Any]]) -> User: # noqa
         if isinstance(obj_in, dict):
             update_data = obj_in
         else:
@@ -39,39 +39,39 @@ async def update(self, db: AgnosticDatabase, *, db_obj: User, obj_in: Union[User
             update_data["email_validated"] = False
         return await super().update(db, db_obj=db_obj, obj_in=update_data)
 
-    async def authenticate(self, db: AgnosticDatabase, *, email: str, password: str) -> Optional[User]:
+    async def authenticate(self, db: AgnosticDatabase, *, email: str, password: str) -> User | None: # noqa
         user = await self.get_by_email(db, email=email)
         if not user:
             return None
-        if not verify_password(plain_password=password, hashed_password=user.hashed_password):
+        if not verify_password(plain_password=password, hashed_password=user.hashed_password): # noqa
             return None
         return user
 
-    async def validate_email(self, db: AgnosticDatabase, *, db_obj: User) -> User:
+    async def validate_email(self, db: AgnosticDatabase, *, db_obj: User) -> User: # noqa
         obj_in = UserUpdate(**UserInDB.model_validate(db_obj).model_dump())
         obj_in.email_validated = True
         return await self.update(db=db, db_obj=db_obj, obj_in=obj_in)
 
-    async def activate_totp(self, db: AgnosticDatabase, *, db_obj: User, totp_in: NewTOTP) -> User:
+    async def activate_totp(self, db: AgnosticDatabase, *, db_obj: User, totp_in: NewTOTP) -> User: # noqa
         obj_in = UserUpdate(**UserInDB.model_validate(db_obj).model_dump())
         obj_in = obj_in.model_dump(exclude_unset=True)
         obj_in["totp_secret"] = totp_in.secret
         return await self.update(db=db, db_obj=db_obj, obj_in=obj_in)
 
-    async def deactivate_totp(self, db: AgnosticDatabase, *, db_obj: User) -> User:
+    async def deactivate_totp(self, db: AgnosticDatabase, *, db_obj: User) -> User: # noqa
         obj_in = UserUpdate(**UserInDB.model_validate(db_obj).model_dump())
         obj_in = obj_in.model_dump(exclude_unset=True)
         obj_in["totp_secret"] = None
         obj_in["totp_counter"] = None
         return await self.update(db=db, db_obj=db_obj, obj_in=obj_in)
 
-    async def update_totp_counter(self, db: AgnosticDatabase, *, db_obj: User, new_counter: int) -> User:
+    async def update_totp_counter(self, db: AgnosticDatabase, *, db_obj: User, new_counter: int) -> User:  # noqa
         obj_in = UserUpdate(**UserInDB.model_validate(db_obj).model_dump())
         obj_in = obj_in.model_dump(exclude_unset=True)
         obj_in["totp_counter"] = new_counter
         return await self.update(db=db, db_obj=db_obj, obj_in=obj_in)
 
-    async def toggle_user_state(self, db: AgnosticDatabase, *, obj_in: Union[UserUpdate, Dict[str, Any]]) -> User:
+    async def toggle_user_state(self, db: AgnosticDatabase, *, obj_in: Union[UserUpdate, Dict[str, Any]]) -> User: # noqa
         db_obj = await self.get_by_email(db, email=obj_in.email)
         if not db_obj:
             return None

{{cookiecutter.project_slug}}/backend/app/app/main.py

@@ -23,8 +23,9 @@ async def app_init(app: FastAPI):
     app.add_middleware(
         CORSMiddleware,
         # Trailing slash causes CORS failures from these supported domains
-        allow_origins=[str(origin).rstrip("/") for origin in settings.BACKEND_CORS_ORIGINS],
+        allow_origins=[str(origin).strip("/") for origin in settings.BACKEND_CORS_ORIGINS], # noqa
         allow_credentials=True,
         allow_methods=["*"],
         allow_headers=["*"],
     )
+

{{cookiecutter.project_slug}}/backend/app/app/models/user.py

@@ -1,5 +1,5 @@
 from __future__ import annotations
-from typing import TYPE_CHECKING, Any, Optional
+from typing import TYPE_CHECKING, Any
 from datetime import datetime
 from pydantic import EmailStr
 from odmantic import ObjectId, Field
@@ -21,7 +21,7 @@ class User(Base):
     email: EmailStr
     hashed_password: Any = Field(default=None)
     totp_secret: Any = Field(default=None)
-    totp_counter: Optional[int] = Field(default=None)
+    totp_counter: int | None = Field(default=None)
     email_validated: bool = Field(default=False)
     is_active: bool = Field(default=False)
     is_superuser: bool = Field(default=False)

{{cookiecutter.project_slug}}/backend/app/app/schemas/base_schema.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 from pydantic import ConfigDict, BaseModel, Field
-from typing import Optional
 from uuid import UUID
 from datetime import date, datetime
 import json
@@ -11,7 +10,7 @@
 class BaseSchema(BaseModel):
     @property
     def as_db_dict(self):
-        to_db = self.model_dump(exclude_defaults=True, exclude_none=True, exclude={"identifier", "id"})
+        to_db = self.model_dump(exclude_defaults=True, exclude_none=True, exclude={"identifier", "id"}) # noqa
         for key in ["id", "identifier"]:
             if key in self.model_dump().keys():
                 to_db[key] = self.model_dump()[key].hex
@@ -21,15 +20,15 @@ def as_db_dict(self):
 class MetadataBaseSchema(BaseSchema):
     # Receive via API
     # https://www.dublincore.org/specifications/dublin-core/dcmi-terms/#section-3
-    title: Optional[str] = Field(None, description="A human-readable title given to the resource.")
-    description: Optional[str] = Field(
+    title: str | None = Field(None, description="A human-readable title given to the resource.") # noqa
+    description: str | None = Field(
         None,
         description="A short description of the resource.",
     )
-    isActive: Optional[bool] = Field(default=True, description="Whether the resource is still actively maintained.")
-    isPrivate: Optional[bool] = Field(
+    isActive: bool | None = Field(default=True, description="Whether the resource is still actively maintained.") # noqa
+    isPrivate: bool | None = Field(
         default=True,
-        description="Whether the resource is private to team members with appropriate authorisation.",
+        description="Whether the resource is private to team members with appropriate authorisation.",  # noqa
     )
 
 

{{cookiecutter.project_slug}}/backend/app/app/schemas/token.py

@@ -1,11 +1,10 @@
-from typing import Optional
 from pydantic import BaseModel, ConfigDict, SecretStr
 from odmantic import Model, ObjectId
 
 
 class RefreshTokenBase(BaseModel):
     token: SecretStr
-    authenticates: Optional[Model] = None
+    authenticates: Model | None = None
 
 
 class RefreshTokenCreate(RefreshTokenBase):
@@ -22,19 +21,19 @@ class RefreshToken(RefreshTokenUpdate):
 
 class Token(BaseModel):
     access_token: SecretStr
-    refresh_token: Optional[SecretStr] = None
+    refresh_token: SecretStr | None = None
     token_type: str
 
 
 class TokenPayload(BaseModel):
-    sub: Optional[ObjectId] = None
-    refresh: Optional[bool] = False
-    totp: Optional[bool] = False
+    sub: ObjectId | None = None
+    refresh: bool | None = False
+    totp: bool | None = False
 
 
 class MagicTokenPayload(BaseModel):
-    sub: Optional[ObjectId] = None
-    fingerprint: Optional[ObjectId] = None
+    sub: ObjectId | None = None
+    fingerprint: ObjectId | None = None
 
 
 class WebToken(BaseModel):

{{cookiecutter.project_slug}}/backend/app/app/schemas/totp.py

@@ -1,14 +1,13 @@
-from typing import Optional
 from pydantic import BaseModel, SecretStr
 
 
 class NewTOTP(BaseModel):
-    secret: Optional[SecretStr] = None
+    secret: SecretStr | None = None
     key: str
     uri: str
 
 
 class EnableTOTP(BaseModel):
     claim: str
     uri: str
-    password: Optional[SecretStr] = None
+    password: SecretStr | None = None

{{cookiecutter.project_slug}}/backend/app/app/schemas/user.py

@@ -1,4 +1,3 @@
-from typing import Optional
 from typing_extensions import Annotated
 from pydantic import BaseModel, ConfigDict, Field, EmailStr, StringConstraints, field_validator, SecretStr
 from odmantic import ObjectId
@@ -11,27 +10,27 @@ class UserLogin(BaseModel):
 
 # Shared properties
 class UserBase(BaseModel):
-    email: Optional[EmailStr] = None
-    email_validated: Optional[bool] = False
-    is_active: Optional[bool] = True
-    is_superuser: Optional[bool] = False
+    email: EmailStr | None = None
+    email_validated: bool | None = False
+    is_active: bool | None = True
+    is_superuser: bool | None = False
     full_name: str = ""
 
 
 # Properties to receive via API on creation
 class UserCreate(UserBase):
     email: EmailStr
-    password: Optional[Annotated[str, StringConstraints(min_length=8, max_length=64)]] = None
+    password: Annotated[str | None, StringConstraints(min_length=8, max_length=64)] = None # noqa
 
 
 # Properties to receive via API on update
 class UserUpdate(UserBase):
-    original: Optional[Annotated[str, StringConstraints(min_length=8, max_length=64)]] = None
-    password: Optional[Annotated[str, StringConstraints(min_length=8, max_length=64)]] = None
+    original: Annotated[str | None, StringConstraints(min_length=8, max_length=64)] = None # noqa
+    password: Annotated[str | None, StringConstraints(min_length=8, max_length=64)] = None  # noqa
 
 
 class UserInDBBase(UserBase):
-    id: Optional[ObjectId] = None
+    id: ObjectId | None = None
     model_config = ConfigDict(from_attributes=True)
 
 
@@ -56,6 +55,6 @@ def evaluate_totp_secret(cls, totp_secret):
 
 # Additional properties stored in DB
 class UserInDB(UserInDBBase):
-    hashed_password: Optional[SecretStr] = None
-    totp_secret: Optional[SecretStr] = None
-    totp_counter: Optional[int] = None
+    hashed_password: SecretStr | None = None
+    totp_secret: SecretStr | None = None
+    totp_counter: int | None = None