Merge pull request 'add updates to show bedir' (#14) from another into main

Reviewed-on: #14

auth/models.py

@@ -1,135 +1,27 @@
-from enum import Enum
+from sqlalchemy import Column, Integer, String, Enum, DateTime
-from backend.config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES
+import enum, datetime
-from backend.config import pwd_context, get_session_db
-from datetime import datetime, timedelta, timezone
-from pydantic import BaseModel
-from fastapi import Depends, HTTPException
-from typing import Annotated, Optional
-from fastapi.security import OAuth2PasswordBearer
-from passlib.context import CryptContext
-import jwt
-from sqlmodel import SQLModel, Field, Session, select
-from pydantic.networks import EmailStr
 
-class Token(BaseModel):
+from ..config import Base
-    access_token: str
-    token_type: str
 
-
+class Role(str, enum.Enum):
-### ENUMS ###
-class Role(str, Enum):
-    user = "user"
     admin = "admin"
-    guest = "guest"
+    user = "user"
     mod = "mod"
 
-class Status(str, Enum):
+class Status(str, enum.Enum):
-    active = "active"
     banned = "banned"
+    active = "active"
     suspended = "suspended"
 
-### KULLANICI MODELLERİ ###
+class User(Base):
-class UserBase(SQLModel):
+    __tablename__ = "users"
-    username: Optional[str] = None
+    user_id = Column(Integer, primary_key=True)
-    user_id: Optional[int] = None
+    username = Column(String, unique=True)
-    role: Optional[Role] = None
+    name = Column(String)
-    status: Optional[Status] = None
+    surname = Column(String)
-
+    hashedPassword = Column(String)
-class UserInDb(UserBase):
+    email = Column(String, unique=True)
-    hashed_password: str | None = None
+    role = Column(Enum(Role), default=Role.user)
-
+    status = Column(Enum(Status), default=Status.active)
-class UserPublic(UserBase):
+    bio = Column(String(144))
-    pass
+    created_date = Column(DateTime, default=datetime.datetime.utcnow)
-
-class UserCreate(BaseModel):
-    username: Optional[str] = None
-    role: Optional[Role] = None
-    email : EmailStr | None = None
-    status: Optional[Status] = None
-    password : str  | None = None
-
-### VERİTABANI MODELİ ###
-class DBUser(SQLModel, table=True):
-    __tablename__ = "users"  # opsiyonel, sqlmodel bunu otomatik de atar
-    user_id: Optional[int] = Field(default=None, primary_key=True)
-    username: str = Field(index=True, nullable=False)
-    hashed_password: str = Field(nullable=False)
-    role: Role = Field(default=Role.user)
-    status: Status = Field(default=Status.active)
-
-
-### AUTH ###
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
-
-
-def verify_password(plain_password: str, hashed_password: str) -> bool:
-    return pwd_context.verify(plain_password, hashed_password)
-
-def get_password_hash(password: str) -> str:
-    return pwd_context.hash(password)
-
-def authenticate_user(
-        session: Annotated[Session, Depends(get_session_db)],
-        username: str,
-        password: str
-    ) -> UserInDb | None:
-
-    statement = select(DBUser).where(DBUser.username == username)
-    result = session.exec(statement).first()
-    if not result or not verify_password(password, result.hashed_password):
-        return None
-    return result
-
-
-def create_access_token(
-        data: dict,
-        expires_delta: Optional[timedelta] = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
-) -> str:
-    to_encode = data.copy()
-    expire = datetime.now(timezone.utc) + expires_delta
-    to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm="HS256")
-    return encoded_jwt
-
-
-async def get_user(
-        session: Annotated[Session, Depends(get_session_db)],
-        username: str
-    ) -> UserInDb | None:
-
-    statement = select(DBUser).where(DBUser.username == username)
-    result = session.exec(statement).first()
-    return result
-
-
-async def get_current_user(
-    token: Annotated[str, Depends(oauth2_scheme)],
-    session: Annotated[Session, Depends(get_session_db)]
-) -> UserPublic:
-
-    credentials_exception = HTTPException(
-        status_code=401,
-        detail="Invalid credentials",
-        headers={"WWW-Authenticate": "Bearer"},
-    )
-    try:
-        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
-        username: Optional[str] = payload.get("sub")
-        if username is None:
-            raise credentials_exception
-    except jwt.PyJWTError:
-        raise credentials_exception
-
-    user = await get_user(session, username)
-    if user is None:
-        raise credentials_exception
-    return user
-
-
-async def get_current_active_user(
-    current_user: Annotated[UserInDb, Depends(get_current_user)]
-) -> UserPublic:
-
-    if current_user.status == Status.banned:
-        raise HTTPException(status_code=400, detail="Inactive user")
-    return current_user

auth/router.py

@@ -1,15 +1,14 @@
-from fastapi import APIRouter, Depends, HTTPException, status
+import os
-from .models import Token, UserPublic
-from .models import authenticate_user, create_access_token
-from datetime import timedelta
-from ..auth.models import get_password_hash, verify_password
-from typing import Annotated
-from sqlmodel import Session
-from ..config import get_session_db
-from fastapi import Depends
-from fastapi.security import OAuth2PasswordRequestForm
-from .models import UserCreate, DBUser
 
+from fastapi import APIRouter, HTTPException
+import bcrypt
+import jwt
+
+from fastapi import Depends
+from sqlalchemy.orm import Session
+from .models import User
+from .schemas import UserCreate, UserOut, UserLogin
+from ..config import get_db
 
 router = APIRouter(
     prefix="/auth",
@@ -18,41 +17,40 @@ router = APIRouter(
     dependencies=[],
 )
 
-@router.post('/login')
+def create_token(user: User):
-async def login_for_access_token(
+    return jwt.encode({"sub": user.username}, os.getenv("SECRET_KEY"), algorithm=os.getenv("ALGORITHM"))
-    form_data : Annotated[OAuth2PasswordRequestForm, Depends()],
-    session : Annotated[Session, Depends(get_session_db)],
-) -> Token:
 
-    user = authenticate_user(session, form_data.username, form_data.password)
+def verify_token(token: str):
-    if not user:
+    try:
-        raise HTTPException(
+        data = jwt.decode(token, os.getenv("SECRET_KEY"), algorithms=[os.getenv("ALGORITHM")])
-            status_code=status.HTTP_401_UNAUTHORIZED,
+        return data.get("sub")
-            detail="Incorrect username or password",
+    except jwt.ExpiredSignatureError:
-            headers={"WWW-Authenticate": "Bearer"},
+        raise HTTPException(401, "Token expired")
-        )
+    except jwt.InvalidTokenError:
-    access_token_expires = timedelta(minutes=30)
+        raise HTTPException(401, "Invalid token")
-    access_token = create_access_token(
-        data={"sub": user.username, "role": user.role, 'status': user.status}, expires_delta=access_token_expires
-    )
-    return Token(access_token=access_token, token_type="bearer")
 
 
-@router.post('/register', response_model=UserPublic)
+@router.post("/register")
-async def create_user(
+def register(user: UserCreate, db: Session = Depends(get_db)):
-        session : Annotated[Session, Depends(get_session_db)],
+    if db.query(User).filter_by(username=user.username).first():
-        user : Annotated[UserCreate, Depends()]
+        raise HTTPException(400, "Username taken")
-):
+    hashed = bcrypt.hashpw(user.password.encode(), bcrypt.gensalt()).decode()
-    user_dict = user.dict()
+    db_user = User(**user.model_dump(exclude={"password"}), hashedPassword=hashed)
-    print(user.password)
+    db.add(db_user)
-    user_dict['hashed_password'] = get_password_hash(user.password)
+    db.commit()
-    print (user_dict['hashed_password'])
+    return {"msg": "User created"}
 
-    if not verify_password(user.password, user_dict['hashed_password']):
+@router.post("/login")
-        raise HTTPException(status_code=400, detail="Password hashing failed")
+def login(user: UserLogin, db: Session = Depends(get_db)):
+    db_user = db.query(User).filter_by(username=user.username).first()
+    if not db_user or not bcrypt.checkpw(user.password.encode(), db_user.hashedPassword.encode()):
+        raise HTTPException(401, "Invalid creds")
+    return {"token": create_token(db_user)}
 
-    db_user = DBUser.model_validate(user_dict)
+@router.get("/me", response_model=UserOut)
-    session.add(db_user)
+def get_me(token: str, db: Session = Depends(get_db)):
-    session.commit()
+    username = verify_token(token)
-    session.refresh(db_user)
+    if not username:
-    return db_user
+        raise HTTPException(401, "Invalid token")
+    user = db.query(User).filter_by(username=username).first()
+    return user

auth/schemas.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel, EmailStr
+from .models import Role, Status
+
+class UserCreate(BaseModel):
+    username: str
+    name: str
+    surname: str
+    password: str
+    email: EmailStr
+    bio: str = ""
+
+class UserOut(BaseModel):
+    username: str
+    name: str
+    surname: str
+    email: EmailStr
+    role: Role
+    status: Status
+    bio: str
+
+class UserLogin(BaseModel):
+    username: str
+    password: str

config.py

@@ -1,17 +1,13 @@
 from sqlalchemy import create_engine
-from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker, declarative_base
-from sqlalchemy.orm import sessionmaker
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-from passlib.context import CryptContext
-from sqlmodel import SQLModel, Field, Session
 from dotenv import load_dotenv
 import os
 
 load_dotenv()
 
 
-# Veritabanı URL'sini oluştur
 DATABASE_URL = (
     f"postgresql://{os.getenv('USERNAME_DB')}:"
     f"{os.getenv('PASSWORD_DB')}@"
@@ -20,16 +16,24 @@ DATABASE_URL = (
     f"{os.getenv('NAME_DB')}"
 )
 
-engine = create_engine(DATABASE_URL, echo=False)
-def init_db():
-    SQLModel.metadata.create_all(engine)
 
-def get_session_db():
+engine = create_engine(DATABASE_URL)
-    with Session(engine) as session:
+SessionLocal = sessionmaker(bind=engine)
-        yield session
+Base = declarative_base()
+
+from .auth.models import *
+from .items.models import *
+
+Base.metadata.create_all(bind=engine)
 
 
-### SECRET KEY ###
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
 origins = [
     "http://localhost",
     "http://localhost:8080",
@@ -38,10 +42,6 @@ origins = [
 ]
 
 app = FastAPI()
-@app.on_event("startup")
-def on_startup():
-    init_db()
-
 app.add_middleware(
     CORSMiddleware,
     allow_origins=origins,
@@ -49,6 +49,3 @@ app.add_middleware(
     allow_methods=["*"],
     allow_headers=["*"],
 )
-
-
-

items/models.py

@@ -1,12 +1,15 @@
-from datetime import datetime, timedelta, timezone
+from datetime import datetime
-from ..auth.models import UserBase
+from ..config import Base
-
+from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
-class UserProfile(UserBase):
+from sqlalchemy.orm import relationship, Mapped, mapped_column
-    bio : str | None = None
-    created_date : datetime | None = None
-    collections : list[str] | None = None
-    items :list[str] | None = None
-
-
-
 
+class Item(Base):
+    __tablename__ = "items"
+    item_id = Column(Integer, primary_key=True)
+    name = Column(String(100), nullable=False)
+    description = Column(String(500), nullable=True)
+    price = Column(Integer, nullable=False)
+    created_date = Column(DateTime, default=datetime.utcnow)
+    updated_date = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
+    user: Mapped["User"] = relationship(back_populates="items")

items/router.py

@@ -1,7 +1,4 @@
-from .models import UserProfile
 from fastapi import APIRouter, Depends
-from typing import Annotated
-from ..auth.models import get_current_active_user
 
 router = APIRouter(
     prefix="/items",
@@ -10,9 +7,7 @@ router = APIRouter(
     dependencies=[],
 )
 
-@router.get('/profile', response_model=UserProfile)
+
-async def get_user_profile(
+@router.get("/")
-    current_user: Annotated[UserProfile, Depends(get_current_active_user)]
+async def get_items():
-) -> UserProfile:
+    return {"message": "List of items"}
-    
-    return current_user