Basic user model and login authentication :]

app/config.py

@@ -1,8 +1,8 @@
 from pydantic_settings import BaseSettings
 
-class Settings(BaseSettings):
 
-    app_name: str = "DagemarkBackend"
+class Settings(BaseSettings):
+    app_name: str = "Backend"
 
     app_version: str = "0.1"
 
@@ -10,7 +10,14 @@ class Settings(BaseSettings):
 
     debug: bool = True
 
+    secret_key: str = ""
+
+    algorithm: str = ""
+
+    access_token_expire_minutes: int = 10
+
     class Config:
         env_file = ".env"
 
-settings = Settings()
+
+settings = Settings()

app/logic/user_logic.py

@@ -0,0 +1,90 @@
+from datetime import UTC, datetime, timedelta
+from typing import Annotated
+
+import jwt
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from pwdlib import PasswordHash
+
+from app.config import Settings
+from app.database import get_from_database_where
+from app.models.token import TokenData
+from app.models.user import User, UserInDB
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+password_hash = PasswordHash.recommended()
+
+DUMMY_HASH = password_hash.hash("testpassword")
+
+
+def verify_password(plain_password: str, hashed_password: str):
+    return password_hash.verify(plain_password, hashed_password)
+
+
+def get_password_hash(password: str):
+    return password_hash.hash(password)
+
+
+def get_user(username: str):
+    user_dict = get_from_database_where(User, User.username, username, True)
+    if user_dict is not None:
+        return UserInDB(**user_dict)
+
+
+def authenticate_user(username: str, password: str):
+    user = get_user(username)
+    if not user:
+        verify_password(password, DUMMY_HASH)
+        return False
+    if not verify_password(password, user.hashed_password):
+        return False
+    return user
+
+
+def create_access_token(
+    data: dict[str, str],
+    expires_delta: timedelta | None = None,
+):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.now(UTC) + expires_delta
+    else:
+        expire = datetime.now(UTC) + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(
+        to_encode,
+        Settings.secret_key,
+        algorithm=Settings.algorithm,
+    )
+    return encoded_jwt
+
+
+async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(
+            token,
+            Settings.secret_key,
+            algorithms=[Settings.algorithm],
+        )
+        username = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+        token_data = TokenData(username=username)
+    except jwt.InvalidTokenError:
+        raise credentials_exception
+    user = get_user(token_data.username)  # type: ignore Look just above here idiot!
+    if user is None:
+        raise credentials_exception
+    return user
+
+
+async def get_current_active_user(
+    current_user: Annotated[User, Depends(get_current_user)],
+):
+    return current_user

app/models/token.py

@@ -0,0 +1,10 @@
+from sqlmodel import SQLModel
+
+
+class Token(SQLModel):
+    access_token: str
+    token_type: str
+
+
+class TokenData(SQLModel):
+    username: str | None = None

app/models/user.py

@@ -0,0 +1,13 @@
+from sqlmodel import Field, SQLModel
+
+
+class User(SQLModel, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+
+    username: str
+
+    authority: int
+
+
+class UserInDB(User):
+    hashed_password: str

app/routers/user_route.py

@@ -0,0 +1,45 @@
+from datetime import timedelta
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordRequestForm
+
+from app.config import Settings
+from app.logic.user_logic import (
+    authenticate_user,
+    create_access_token,
+    get_current_user,
+)
+from app.models.token import Token
+from app.models.user import User
+
+router = APIRouter(prefix="/users", tags=["users"])
+
+
+@router.get("/")
+def get_default():
+    return {"message": "Root for Users API"}
+
+
+@router.get("/me")
+async def read_me(current_user: Annotated[User, Depends(get_current_user)]):
+    return current_user
+
+
+@router.post("/token")
+async def login_for_access_token(
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
+) -> Token:
+    user = authenticate_user(form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token_expires = timedelta(minutes=Settings.access_token_expire_minutes)
+    access_token = create_access_token(
+        data={"sub": user.username},
+        expires_delta=access_token_expires,
+    )
+    return Token(access_token=access_token, token_type="bearer")