zwischenstand

backend/src/backend/internal/authentication.py

@@ -0,0 +1,35 @@
+from sqlmodel import select
+from backend.models.user import User
+from backend.internal.database import get_session
+from sqlmodel import Session
+from fastapi import HTTPException
+from fastapi.security.oauth2 import OAuth2PasswordBearer
+from fastapi import Depends
+from datetime import datetime
+from datetime import timedelta
+
+
+SECRET_KEY = "secret-key"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_DAYS = 360
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+def create_access_token(data: dict, expires_delta: timedelta | None = None):
+    to_encode = data.copy()
+    expire = datetime.utcnow() + (expires_delta or timedelta(days=ACCESS_TOKEN_EXPIRE_DAYS))
+    to_encode.update({"exp": expire})
+    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+
+async def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)):
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        user_code: str = payload.get("sub")
+        if user_code is None:
+            raise HTTPException(status_code=401, detail="Invalid token")
+    except JWTError:
+        raise HTTPException(status_code=401, detail="Invalid token")
+    user = session.exec(select(User).where(User.user_code == user_code))
+    if user is None:
+        raise HTTPException(status_code=401, detail="User not found")
+    return user

backend/src/backend/internal/user_code_gen.py

@@ -1,4 +1,5 @@
 import string
 import random
+
 def generate_user_code() -> str:
-    return ''.join(random.choices(string.ascii_letters, k=4)).upper()
+    return ''.join(random.choices(string.ascii_uppercase, k=4))

backend/src/backend/models/comment.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from datetime import datetime
 import uuid
 from typing import TYPE_CHECKING
@@ -11,10 +12,11 @@ if TYPE_CHECKING:
 class CommentBase(SQLModel):
     pass
 
+@dataclass
 class Comment(CommentBase, table=True):
-    id: uuid.UUID | None = Field(default_factory=uuid.uuid4, primary_key=True)
+    id: uuid.UUID | None = Field(default_factory=uuid.uuid7, primary_key=True)
     slug: str = Field(index=True)
     content: str = Field()
-    date_time: datetime = Field(default=datetime.now())
+    date_time: datetime = Field(default_factory=datetime.now)
     user_id: uuid.UUID = Field(foreign_key="user.id")
     user: "User" = Relationship(back_populates="comments")

backend/src/backend/models/user.py

@@ -1,5 +1,6 @@
+from uuid import uuid7
+from datetime import datetime
 from backend.internal.user_code_gen import generate_user_code
-from dataclasses import dataclass
 import uuid
 from typing import TYPE_CHECKING
 from sqlmodel import Field
@@ -13,15 +14,40 @@ class UserBase(SQLModel):
     pass
 
 class User(UserBase, table=True):
-    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+    id: uuid.UUID = Field(default_factory=uuid.uuid7, primary_key=True)
     email: str = Field(unique=True)
     name: str = Field(unique=True)
     user_code: str = Field(default_factory=generate_user_code,unique=True)
     subscription: bool = Field()
+    created_at: datetime = Field(default_factory=datetime.now)
+
     comments: list["Comment"] = Relationship(back_populates="user")
 
-@dataclass
+    def factory(self,
+        id: uuid.UUID = uuid.uuid7(),
+        email: str = "Email@Test.mail",
+        name: str = "TestName",
+        user_code: str = "WYZZ",
+        subscription: bool = False,
+        ) -> User:
+        self.id=id
+        self.email=email
+        self.name=name
+        self.user_code=user_code
+        self.subscription=subscription
+        return self
+
 class NewUser(UserBase):
     email: str
     name: str
     subscription: bool
+
+    def factory(self,
+        email: str = "Email@Test.mail",
+        name: str = "TestName",
+        subscription: bool = False
+    ) -> NewUser:
+        self.email = email
+        self.name = name
+        self.subscription = subscription
+        return self

backend/src/backend/routers/test_users.py

@@ -39,7 +39,7 @@ def test_register_user_saved(session:Session, client: TestClient):
         "subscription": True
     }
     response = client.post(url="/users/", json=new_user)
-    assert response.status_code == 200
+    assert response.status_code == 201
 
     db_user = session.exec(select(User).where(User.email == "Test@Test.test")).first()
     assert db_user
@@ -48,19 +48,37 @@ def test_register_user_saved(session:Session, client: TestClient):
     assert db_user.subscription == True
 
 def test_email_not_unique_returns_400(session: Session, client: TestClient):
-    new_user = NewUser(email="Not@Unique.de",name="Tim", subscription=True)
+    new_user = NewUser(email="NotUnique",name="Müller", subscription=True) # TODO: durch factory ersetzen
 
-    user = User.model_validate(new_user.model_copy(update={'name':'Different'}))
+    user = User().factory(email="NotUnique")
     session.add(user)
     session.commit()
     response = client.post(url="/users/", json=new_user.model_dump())
     assert response.status_code == 400
 
 def test_name_not_unique_returns_400(session: Session, client: TestClient):
-    new_user = NewUser(email="Absolutly@Unique.de",name="Chiara", subscription=True)
+    new_user = NewUser(email="wewqe@2ewew",name="Chiara", subscription=True) # TODO: durch factory ersetzen
 
-    user = User.model_validate(new_user.model_copy(update={'email':'Different@Different.de'}))
+    user = User().factory(name="Chiara")
     session.add(user)
     session.commit()
     response = client.post(url="/users/", json=new_user.model_dump())
-    assert response.status_code == 400
+    assert response.status_code == 400
+
+def test_login_successful(session: Session, client: TestClient):
+    user = User().factory()
+    session.add(user)
+    session.commit()
+    session.refresh(user)
+
+    payload = {
+        "user_code": user.user_code
+    }
+
+    response = client.post(
+        url="/users/login", 
+        data=payload
+    )
+
+    assert response.status_code == 200
+    assert response.json() == "token"

backend/src/backend/routers/users.py

@@ -1,13 +1,5 @@
-# CreateUser <- mail, Name
+from fastapi import Form
-# Login Rate Limit! -> token for everything else
+from sqlmodel import select
-# Forgot Code <- email (rate limit)
-# Delete User <- token
-# change mail <- token, new mail
-# subscribe
-# unsubscribe
-
-# delete all deactivated users after 48h
-
 from starlette.status import HTTP_400_BAD_REQUEST
 from fastapi import HTTPException
 from sqlalchemy.exc import IntegrityError
@@ -25,13 +17,32 @@ router = APIRouter(
     responses={404: {"description": "Not found"}},
 )
 
-@router.post("/")
+@router.post("/", status_code=201)
 def register_user(new_user: NewUser, session: Session = Depends(get_session)):
     user = User.model_validate(new_user)
     
     try:
         session.add(user)
         session.commit()
+        # TODO: send mail
     except IntegrityError as e:
         session.rollback()
         raise HTTPException(HTTP_400_BAD_REQUEST)
+
+@router.post("/login") # TODO: rate limit
+def login(user_code: str = Form(), session: Session = Depends(get_session)) -> str:
+    user = session.exec(select(User).where(User.user_code == user_code)).one   
+
+    if not user:
+        raise HTTPException()     
+    return "token"
+
+
+# Login Rate Limit! -> token for everything else
+# Forgot Code <- email (rate limit)
+# Delete User <- token
+# change mail <- token, new mail
+# subscribe
+# unsubscribe
+
+# delete all deactivated users after 48h