basic users model and views

users/models.py

@@ -1,16 +1,21 @@
-# from sqlmodel import SQLModel, Field
+from sqlalchemy import Column, String, DateTime
-from sqlalchemy import Column, String, DateTime, delete
+from sqlalchemy import update as sqlalchemy_update
-from sqlalchemy import update as sqlalchemy_update, delete as sqlalchemy_delete
 from sqlalchemy.future import select
-from app.db import Base, db
+from app.database.models import BaseCRUD
+from app.database.db import async_session
 from datetime import datetime
 
 from uuid import uuid4
+from passlib.context import CryptContext
 
-class User(Base):
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+class User(BaseCRUD):
     __tablename__ = "users"
-    id = Column(String, primary_key=True)
+    username = Column(String, nullable=False, unique=True)
-    username = Column(String)
+    full_name = Column(String)
+    email = Column(String, nullable=False, unique=True)
+    password = Column(String, nullable=False)
     created_at = Column(DateTime, index=True, default=datetime.utcnow)
 
     def __repr__(self):
@@ -21,57 +26,60 @@ class User(Base):
             f")>"
         )
 
+    def verify_password(self, plain_password):
+        return pwd_context.verify(plain_password, self.password)
+
+    @staticmethod
+    def get_password_hash(password):
+        return pwd_context.hash(password)
+
+    @classmethod
+    async def authenticate_user(cls, username: str, password: str):
+        # TODO: add exception when noone found
+        async with async_session() as db:
+            query = select(cls).where(cls.username == username)
+            users = await db.execute(query)
+            (user,) = users.first()
+
+            if user:
+                if not cls.verify_password(user, password):
+                    return False
+
+            return user
+
     @classmethod
     async def create(cls, **kwargs):
-        user = cls(id=str(uuid4()), **kwargs)
+        if 'plain_password' in kwargs:
-        db.add(user)
+            kwargs['password'] = cls.get_password_hash(kwargs.pop('plain_password'))
-        try:
+        async with async_session() as db:
-            await db.commit()
+            user = cls(id=str(uuid4()), **kwargs)
-        except Exception:
+            db.add(user)
-            await db.rollback()
+            try:
-            raise
+                await db.commit()
-        return user
+                await db.refresh()
+            except Exception:
+                await db.rollback()
+                raise
+            return user
 
     @classmethod
     async def update(cls, id, **kwargs):
-        query = (
+        if 'plain_password' in kwargs:
-            sqlalchemy_update(cls)
+            kwargs['password'] = cls.get_password_hash(kwargs.pop('plain_password'))
-            .where(cls.id == id)
+        async with async_session() as db:
-            .values(**kwargs)
+            query = (
-            .execution_options(synchronize_session="fetch")
+                sqlalchemy_update(cls)
-        )
+                .where(cls.id == id)
-        await db.execute(query)
+                .values(**kwargs)
-        try:
+                .execution_options(synchronize_session="fetch")
-            await db.commit()
+            )
-        except Exception:
+            await db.execute(query)
-            await db.rollback()
+            try:
-            raise
+                await db.commit()
-        return await cls.get(id)
+            except Exception:
-
+                await db.rollback()
-    @classmethod
+                raise
-    async def get(cls, id):
+            return await cls.get(id)
-        query = select(cls).where(cls.id == id)
-        users = await db.execute(query)
-        (user,) = users.first()
-        return user
-
-    @classmethod
-    async def get_all(cls):
-        query = select(cls)
-        users = await db.execute(query)
-        users = users.scalars().all()
-        return users
-
-    @classmethod
-    async def delete(cls, id):
-        query = sqlalchemy_delete(cls).where(cls.id == id)
-        await db.execute(query)
-        try:
-            await db.commit()
-        except Exception:
-            await db.rollback()
-            raise
-        return True
 
 # class UserBase(SQLModel):
 #     username: str

users/views.py

@@ -3,13 +3,8 @@ from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
 from starlette.status import HTTP_401_UNAUTHORIZED
 
-from sqlalchemy.orm import Session
-from sqlalchemy import select
-
-from passlib.context import CryptContext
 from jose import JWTError, jwt
 
-# from app.db import get_session
 from app.settings import settings
 from pydantic import BaseModel
 from typing import List
@@ -19,97 +14,56 @@ from users.models import User
 ALGORITHM = "HS256"
 ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/users/token")
 
-# router = APIRouter()
 
-pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
 
-# def verify_password(plain_password, hashed_password):
+    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
-#     return pwd_context.verify(plain_password, hashed_password)
 
-# def get_password_hash(password):
+    return encoded_jwt
-#     return pwd_context.hash(password)
 
-# def authenticate_user(username: str, password: str, db: Session):
+def verify_access_token(token: str) -> bool:
-#     statement = select(User).where(User.username == username)
+    try:
-#     result = db.execute(statement).fetchone()
+        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
+        user_id: str = payload.get("sub", None)
+        if user_id is None:
+            return False
+    except JWTError:
+        return False
 
-#     if result:
+    return True
-#         print(result.User.__dict__)
-#         if not verify_password(password, result.User.hashed_password):
-#             return False
 
-#     return result.User
-
-# def create_access_token(data: dict, expires_delta: timedelta | None = None):
-#     to_encode = data.copy()
-#     if expires_delta:
-#         expire = datetime.utcnow() + expires_delta
-#     else:
-#         expire = datetime.utcnow() + timedelta(minutes=15)
-#     to_encode.update({"exp": expire})
-#     encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
-
-#     return encoded_jwt
-
-# async def get_current_user(token: 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=[ALGORITHM])
-#         username: str = payload.get("sub", None)
-#         if username is None:
-#             raise credentials_exception
-#         token_data = TokenData(username=username)
-#     except JWTError:
-#         raise credentials_exception
-#     user = get_user(username=token_data.username)
-#     if user is None:
-#         raise credentials_exception
-#     return user
-
-# async def get_current_active_user(current_user: User = Depends(get_current_user)):
-#     if current_user.disabled:
-#         raise HTTPException(status_code=400, detail="Inactive user")
-#     return current_user
-
-# @router.post("/register/", tags=["users"])
-# async def register(db: Session = Depends(get_session)):
-#     pass
-
-# @router.get("/login/", tags=["users"])
-# async def login(session: Session = Depends(get_session)):
-#     username = "johndoe"
-#     statement = select(User).where(User.username == username)
-#     result = session.execute(statement).first()
-
-#     print(result)
-
-#     return {"hi": "there"}
-
-# @router.post("/token", tags=["users"])
-# async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_session)):
-#     user = authenticate_user(form_data.username, form_data.password, db)
-#     if not user:
-#         raise HTTPException(
-#                 status_code=HTTP_401_UNAUTHORIZED,
-#                 detail="Incorrect username or password",
-#                 headers={"WWW-Authenticate": "Bearer"},
-#                 )
-#     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
-#     access_token = create_access_token(
-#             data = {"sub": user.username}, expires_delta=access_token_expires
-#             )
-#     return {"access_token": access_token, "token_type": "bearer"}
 
+async def get_current_user(token: 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=[ALGORITHM])
+        user_id: str = payload.get("sub", None)
+        if user_id is None:
+            raise credentials_exception
+    except JWTError:
+        raise credentials_exception
+    user = await User.get(user_id)
+    if user is None:
+        raise credentials_exception
+    return user
 
 class UserSchema(BaseModel):
     username: str
-
+    full_name: str
+    plain_password: str
+    email: str
 
 class UserSerializer(BaseModel):
     id: str
@@ -118,8 +72,19 @@ class UserSerializer(BaseModel):
     class Config:
             orm_mode = True
 
+class UserPassVerifySchema(BaseModel):
+    id: str
+    plain_password: str
+
+class PasswordSerializer(BaseModel):
+    correct_password: bool
+
+class TokenSchema(BaseModel):
+    token: str
+
 router = APIRouter(
     prefix="/users",
+    tags=["users"],
 )
 
 @router.post("/", response_model=UserSerializer)
@@ -129,6 +94,15 @@ async def create_user(user: UserSchema):
     user = await User.create(**user.dict())
     return user
 
+@router.get("/me", response_model=UserSerializer)
+async def read_users_me(current_user: User = Depends(get_current_user)):
+    return current_user
+
+@router.post("/verify", response_model=PasswordSerializer)
+async def verify_user_password(user: UserPassVerifySchema):
+    us = await User.get(id=user.id)
+    return PasswordSerializer(correct_password=us.verify_password(user.plain_password))
+
 @router.get("/{id}", response_model=UserSerializer)
 async def get_user(id: str):
     user = await User.get(id)
@@ -147,3 +121,23 @@ async def update(id: str, user: UserSchema):
 @router.delete("/{id}", response_model=bool)
 async def delete_user(id: str):
     return await User.delete(id)
+
+@router.post("/check-login", response_model=bool)
+async def check_if_token_still_active(token: TokenSchema):
+    return verify_access_token(token.token)
+
+@router.post("/token")
+async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
+    user = await User.authenticate_user(form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+                status_code=HTTP_401_UNAUTHORIZED,
+                detail="Incorrect username or password",
+                headers={"WWW-Authenticate": "Bearer"},
+                )
+    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = create_access_token(
+            data = {"sub": user.id}, expires_delta=access_token_expires
+            )
+    return {"access_token": access_token, "token_type": "bearer"}
+