From a117f4360c3e80afcb2828609ba82ec2c8f38f9b Mon Sep 17 00:00:00 2001
From: bolexs <boluwatufeadeojo@gmail.com>
Date: Wed, 17 Jul 2024 14:25:02 +0100
Subject: [PATCH] fix remove all imp
---
api/core/dependencies/user.py | 36 -----
api/utils/dict.py | 9 --
api/utils/exceptions.py | 7 -
api/utils/paginator.py | 47 -------
api/utils/string.py | 16 ---
api/utils/utils.py | 13 --
api/v1/models/auth.py | 28 ----
api/v1/routes/auth.py | 253 ----------------------------------
api/v1/schemas/auth.py | 67 ---------
api/v1/services/auth.py | 236 -------------------------------
10 files changed, 712 deletions(-)
delete mode 100644 api/core/dependencies/user.py
delete mode 100644 api/utils/paginator.py
delete mode 100644 api/utils/string.py
delete mode 100644 api/utils/utils.py
delete mode 100644 api/v1/models/auth.py
delete mode 100644 api/v1/routes/auth.py
delete mode 100644 api/v1/schemas/auth.py
delete mode 100644 api/v1/services/auth.py
diff --git a/api/core/dependencies/user.py b/api/core/dependencies/user.py
deleted file mode 100644
index 9f757f3c2..000000000
--- a/api/core/dependencies/user.py
+++ /dev/null
@@ -1,36 +0,0 @@
-from typing import Union
-from fastapi.security import OAuth2PasswordBearer
-from fastapi import Depends, Cookie, HTTPException, status
-from sqlalchemy.orm import Session
-from sqlalchemy.sql import and_
-from jose import JWTError, jwt
-
-from api.v1.schemas import auth as user_schema
-from api.v1.services.auth import User
-from api.v1.models.auth import User as UserModel
-
-from api.db.database import get_db
-from api.core import responses
-
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")
-
-def is_authenticated(
- access_token: Union[str, user_schema.Token] = Depends(oauth2_scheme),
- db: Session = Depends(get_db),
-) -> Union[user_schema.User, JWTError]:
-
- if not access_token:
- raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=responses.INVALID_CREDENTIALS)
-
- userService = User()
-
- access_token_info = userService.verify_access_token(access_token, db)
-
- if type(access_token_info) is JWTError:
- raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=responses.INVALID_CREDENTIALS)
-
- user = userService.fetch(id=access_token_info.id,db=db)
-
- return user
-
-
diff --git a/api/utils/dict.py b/api/utils/dict.py
index 45d1d291b..e69de29bb 100644
--- a/api/utils/dict.py
+++ b/api/utils/dict.py
@@ -1,9 +0,0 @@
-
-def clone_object(obj: dict, unwanted_fields=[]):
- new_obj = {}
- for k in list(obj.keys()):
- if k in unwanted_fields:
- continue
-
- new_obj[k] = obj[k]
- return new_obj
diff --git a/api/utils/exceptions.py b/api/utils/exceptions.py
index e4744ddb1..e69de29bb 100644
--- a/api/utils/exceptions.py
+++ b/api/utils/exceptions.py
@@ -1,7 +0,0 @@
-from fastapi import HTTPException
-
-class CustomException(HTTPException):
-
- @staticmethod
- def PermissionError():
- raise HTTPException(status_code=403, detail="User has no permission to perform this action")
\ No newline at end of file
diff --git a/api/utils/paginator.py b/api/utils/paginator.py
deleted file mode 100644
index f27af595f..000000000
--- a/api/utils/paginator.py
+++ /dev/null
@@ -1,47 +0,0 @@
-from sqlalchemy.orm import Session
-
-def total_row_count(model, organization_id, db: Session):
- return db.query(model).filter(model.organization_id == organization_id).filter(
- model.is_deleted == False).count()
-
-def off_set(page: int, size: int):
- return (page-1)*size
-
-
-def size_validator(size:int):
- if size < 0 or size > 100:
- return "page size must be between 0 and 100"
- return size
-
-
-def page_urls(page: int, size: int, count: int, endpoint: str):
- paging = {}
- if (size + off_set(page, size)) >= count:
- paging['next'] = None
- if page > 1:
- paging['previous'] = f"{endpoint}?page={page-1}&size={size}"
- else:
- paging['previous'] = None
- else:
- paging['next'] = f"{endpoint}?page={page+1}&size={size}"
- if page > 1:
- paging['previous'] = f"{endpoint}?page={page-1}&size={size}"
- else:
- paging['previous'] = None
-
- return paging
-
-
-def build_paginated_response(
- page: int, size: int, total: int, pointers: dict, items
-) -> dict:
- response = {
- "page": page,
- "size": size,
- "total": total,
- "previous_page": pointers["previous"],
- "next_page": pointers["next"],
- "items": items,
- }
-
- return response
\ No newline at end of file
diff --git a/api/utils/string.py b/api/utils/string.py
deleted file mode 100644
index d5daa19b7..000000000
--- a/api/utils/string.py
+++ /dev/null
@@ -1,16 +0,0 @@
-from fastapi import status
-from fastapi.exceptions import HTTPException
-
-def is_empty_string(string: str) -> bool:
- if len(string.strip()) == 0:
- return True
-
- return False
-
-class EmptyStringException(HTTPException):
- def __init__(self, detail: str) -> None:
- super().__init__(
- status_code=status.HTTP_400_BAD_REQUEST,
- detail=detail,
- headers=None,
- )
\ No newline at end of file
diff --git a/api/utils/utils.py b/api/utils/utils.py
deleted file mode 100644
index 0b1a27285..000000000
--- a/api/utils/utils.py
+++ /dev/null
@@ -1,13 +0,0 @@
-def build_paginated_response(
- page: int, size: int, total: int, pointers: dict, items
-) -> dict:
- response = {
- "page": page,
- "size": size,
- "total": total,
- "previous_page": pointers["previous"],
- "next_page": pointers["next"],
- "items": items,
- }
-
- return response
\ No newline at end of file
diff --git a/api/v1/models/auth.py b/api/v1/models/auth.py
deleted file mode 100644
index b85749d7b..000000000
--- a/api/v1/models/auth.py
+++ /dev/null
@@ -1,28 +0,0 @@
-from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, DateTime, BIGINT, Text
-from sqlalchemy.orm import relationship
-from datetime import datetime, date
-from api.db.database import Base
-import passlib.hash as _hash
-
-
-class User(Base):
- __tablename__ = "users"
- id = Column(BIGINT, primary_key=True, autoincrement=True, index=True)
- unique_id = Column(String(255), nullable=True)
- first_name = Column(String(255), nullable=False)
- last_name = Column(String(255), nullable=False)
- email = Column(String(500), unique=True, index=True, nullable=False)
- password = Column(String(500), nullable=False)
- is_active = Column(Boolean, default=True)
- date_created = Column(DateTime,default=datetime.utcnow)
- last_updated = Column(DateTime,default=datetime.utcnow)
- is_deleted = Column(Boolean, default=False)
-
-class BlackListToken(Base):
- __tablename__ = "blacklist_tokens"
- id = Column(BIGINT, primary_key=True, autoincrement=True, index=True)
- created_by = Column(BIGINT, ForeignKey('users.id'), index=True)
- token = Column(String(255), index=True)
- date_created = Column(DateTime, default= datetime.utcnow)
-
-
diff --git a/api/v1/routes/auth.py b/api/v1/routes/auth.py
deleted file mode 100644
index 640880051..000000000
--- a/api/v1/routes/auth.py
+++ /dev/null
@@ -1,253 +0,0 @@
-from fastapi import Depends, Cookie, HTTPException, APIRouter, Depends, status, Response, Request, BackgroundTasks
-from fastapi.security import OAuth2PasswordBearer
-from sqlalchemy.orm import Session
-from datetime import timedelta, datetime
-from typing import Union
-from decouple import config
-from api.v1.schemas import auth as user_schema
-from api.db.database import get_db
-from api.v1.services.auth import User
-from api.v1.models.auth import User as UserModel
-from api.core import responses
-from api.core.dependencies.user import is_authenticated
-
-ACCESS_TOKEN_EXPIRE_MINUTES = int(config('ACCESS_TOKEN_EXPIRE_MINUTES'))
-JWT_REFRESH_EXPIRY = int(config('JWT_REFRESH_EXPIRY'))
-IS_REFRESH_TOKEN_SECURE = True if config('PYTHON_ENV') == "production" else False
-
-
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")
-
-app = APIRouter(tags=["Auth"])
-
-
-@app.post("/signup", status_code=status.HTTP_201_CREATED)
-async def signup(
- response: Response,
- user:user_schema.CreateUser,
- db:Session = Depends(get_db)
-):
-
- """
- Endpoint to create a user
-
- Returns: Created User.
- """
- userService = User()
- created_user = userService.create(user=user, db=db)
-
-
- access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
- access_token = userService.create_access_token(
- data={"id": created_user.id}, db=db, expires_delta=access_token_expires
- )
-
- refresh_token = userService.create_refresh_token(data={"id": created_user.id}, db=db)
-
- response.set_cookie(
- key="refresh_token",
- value=refresh_token,
- max_age=JWT_REFRESH_EXPIRY,
- secure=True,
- httponly=True,
- samesite="strict",
- )
-
- return {"message": responses.SUCCESS,
- "data": user_schema.ShowUser.model_validate(created_user),
- "access_token": access_token,
- "refresh_token": refresh_token,
- "token_type": "bearer"
- }
-
-
-@app.post("/login", status_code=status.HTTP_200_OK)
-async def login_for_access_token(
- response: Response,
- data: user_schema.Login,
- background_task: BackgroundTasks,
- db: Session = Depends(get_db)
-):
- """
- LOGIN
-
- Returns: Logged in User and access token.
- """
-
- userService = User()
- user = userService.authenticate_user(email=data.email, password=data.password, db=db)
- if not user:
- raise HTTPException(
- status_code=status.HTTP_401_UNAUTHORIZED,
- detail=responses.INVALID_CREDENTIALS,
- headers={"WWW-Authenticate": "Bearer"},
- )
- access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
- access_token = userService.create_access_token(
- data={"id": user.id}, db=db, expires_delta=access_token_expires
- )
-
- refresh_token = userService.create_refresh_token(data={"id": user.id}, db=db)
-
- response.set_cookie(
- key="refresh_token",
- value=refresh_token,
- max_age=JWT_REFRESH_EXPIRY,
- secure=True,
- httponly=True,
- samesite="strict",
- path="/"
- )
-
- return {
- "data": user_schema.ShowUser.model_validate(user),
- "access_token": access_token,
- "refresh_token": refresh_token,
- "token_type": "bearer"
- }
-
-
-
-@app.get("/user", status_code=status.HTTP_200_OK)
-async def get_user(
- user: user_schema.User = Depends(is_authenticated),
- db: Session = Depends(get_db),
-):
- """
- Returns an authenticated user information
- """
- print(user)
- return user
-
-
-
-@app.get("/refresh-access-token", status_code=status.HTTP_200_OK)
-async def refresh_access_token(
- response: Response,
- refresh_token: Union[str, None] = Cookie(default=None),
- db: Session = Depends(get_db),
-):
- """Refreshes an access_token with the issued refresh_token
- Parameters
- ----------
- refresh_token : str, None
- The refresh token sent in the cookie by the client (default is None)
-
- Raises
- ------
- UnauthorizedError
- If the refresh token is None.
- """
- print(refresh_token)
- credentials_exception =HTTPException(
- status_code=status.HTTP_401_UNAUTHORIZED,
- detail="Could not validate credentials",
- headers={"WWW-Authenticate": "Bearer"},
- )
-
- if refresh_token is None:
- raise HTTPException(
- detail="Log in to authenticate user",
- status_code=status.HTTP_401_UNAUTHORIZED,
- )
-
- valid_refresh_token = User.verify_refresh_token(
- refresh_token, db
- )
-
- if valid_refresh_token.email is None:
- response.set_cookie(
- key="refresh_token",
- value=refresh_token,
- max_age=JWT_REFRESH_EXPIRY,
- secure=IS_REFRESH_TOKEN_SECURE,
- httponly=True,
- samesite="strict",
- )
-
- print("refresh failed")
- else:
- user = (
- db.query(UserModel)
- .filter(UserModel.id == valid_refresh_token.id)
- .first()
- )
-
- access_token = User.create_access_token(
- {"user_id": valid_refresh_token.id}, db
- )
-
- response.set_cookie(
- key="refresh_token",
- value=refresh_token,
- max_age=JWT_REFRESH_EXPIRY,
- secure=IS_REFRESH_TOKEN_SECURE,
- httponly=True,
- samesite="strict",
- )
-
- # Access token expires in 15 mins,
- return {"user": user_schema.ShowUser.model_validate(user), "access_token": access_token, "expires_in": 900}
-
-
-
-@app.post("/logout", status_code=status.HTTP_200_OK)
-async def logout_user(
- request: Request,
- response: Response,
- user: user_schema.User = Depends(is_authenticated),
- db: Session = Depends(get_db),
-):
- """
- This endpoint logs out an authenticated user.
-
- Returns message: User logged out successfully.
- """
-
- userService = User()
- access_token = request.headers.get('Authorization')
-
- logout = userService.logout(token=access_token, user=user, db=db)
-
- response.set_cookie(
- key="refresh_token",
- max_age="0",
- secure=True,
- httponly=True,
- samesite="strict",
- )
-
- return {"message": "User logged out successfully."}
-
-
-@app.delete("/users/{user_id}")
-async def delete_user(
- user_id: int,
- user: user_schema.User = Depends(is_authenticated),
- db: Session = Depends(get_db),
-):
-
- """
- This endpoint deletes a user from the db. (Soft delete)
-
- Returns message: User deleted successfully.
- """
- userService = User()
- deleted_user = userService.delete(db=db, id=user_id)
-
- return {"message": "User deleted successfully."}
-
-
-@app.post("/users/roles", status_code=status.HTTP_200_OK)
-async def create_user_roles(
- user: user_schema.ShowUser = Depends(is_authenticated),
- db:Session = Depends(get_db)
-):
-
- """
- Endpoint to create custom roles for users mixing permissions.
-
- Returns created role
-
- """
- pass
\ No newline at end of file
diff --git a/api/v1/schemas/auth.py b/api/v1/schemas/auth.py
deleted file mode 100644
index c2cfc8e28..000000000
--- a/api/v1/schemas/auth.py
+++ /dev/null
@@ -1,67 +0,0 @@
-from uuid import uuid4
-from pydantic import BaseModel, model_validator, EmailStr
-from typing import Optional
-from fastapi import HTTPException, status
-from datetime import datetime
-from api.db.database import SessionLocal
-from api.v1.models.auth import User
-from api.core import responses
-
-"""
-TODO: PASSWORD COMPLEXITY VALIDATION ON CREATEUSER SCHEMA
- UNIQUE_ID SHOULD NOT ALREADY EXIST
-
-
-"""
-class Login(BaseModel):
- email: EmailStr
- password: str
-
-class Token(BaseModel):
- access_token: str
- token_type: str
-
-class TokenData(BaseModel):
- id:int
- email:str
-
-class UserBase(BaseModel):
- first_name: str
- last_name: str
- email: str
- unique_id: Optional[str] = None
- is_active: bool = True
- date_created: Optional[datetime] = datetime.utcnow()
- last_updated: Optional[datetime] = datetime.utcnow()
-
- class Config:
- from_attributes = True
-
-
-class CreateUser(UserBase):
- password: str
-
- class Config:
- from_attributes = True
-
-
- #validate email not in use
- @model_validator(mode='before')
- @classmethod
- def validate_email(cls, values):
- email = values.get("email")
-
- with SessionLocal() as db:
- user_email = db.query(User).filter(User.email == email).first()
- if user_email:
- raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail= responses.EMAIL_IN_USE)
-
-
- return values
-
-class ShowUser(UserBase):
- id: int
- is_deleted: Optional[bool]
-
- class Config:
- from_attributes=True
\ No newline at end of file
diff --git a/api/v1/services/auth.py b/api/v1/services/auth.py
deleted file mode 100644
index 16255be29..000000000
--- a/api/v1/services/auth.py
+++ /dev/null
@@ -1,236 +0,0 @@
-import passlib.hash as _hash
-from passlib.context import CryptContext
-from fastapi.security import OAuth2PasswordBearer
-from fastapi import Depends
-from jose import JWTError, jwt
-from sqlalchemy.orm import Session
-from sqlalchemy.sql import or_
-from fastapi import HTTPException, status
-from datetime import datetime,timedelta
-from typing import Annotated, Union
-from uuid import uuid4
-from decouple import config
-
-from api.v1.models.auth import User as UserModel, BlackListToken
-from api.v1.schemas import auth as user_schema
-from api.core import responses
-from api.core.base.services import Service
-
-
-SECRET_KEY = config('SECRET_KEY')
-ALGORITHM = config("ALGORITHM")
-pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
-JWT_REFRESH_EXPIRY = int(config("JWT_REFRESH_EXPIRY"))
-
-
-class User(Service):
-
- def __init__(self) -> None:
- pass
-
- def create(self, user: user_schema.CreateUser, db: Session):
- created_user = UserModel(unique_id=user.unique_id,
- first_name=user.first_name,
- last_name=user.last_name,
- email=user.email,
- date_created=user.date_created,
- last_updated=user.last_updated,
- is_active=user.is_active,
- password=self.hash_password(user.password))
- db.add(created_user)
- db.commit()
-
- return created_user
-
- @staticmethod
- def fetch(db: Session, id: int = None, unique_id: str = None) -> user_schema.User:
- if id is None and unique_id is None:
- raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=responses.ID_OR_UNIQUE_ID_REQUIRED)
-
- user = db.query(UserModel).filter(UserModel.id==id).filter(UserModel.is_deleted==False).first()
- if user is None:
- raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=responses.NOT_FOUND)
-
- return user
-
- @staticmethod
- def fetch_all():
- pass
-
- @staticmethod
- def fetch_by_email(email: str, db: Session) -> user_schema.User:
- user = db.query(UserModel).filter(UserModel.email == email, UserModel.is_deleted==False).first()
-
- return user
-
- def update(self):
- pass
-
- @classmethod
- def delete(cls,db: Session, id: int=None, unique_id: str=None) -> user_schema.User:
- user = cls.fetch(id=id, unique_id=unique_id, db=db)
- user.is_deleted = True
- db.commit()
- return user
-
- @classmethod
- async def get_current_user(cls, token: Annotated[str, Depends(oauth2_scheme)], db:Session) -> user_schema.User:
- credentials_exception = HTTPException(
- status_code=status.HTTP_401_UNAUTHORIZED,
- detail=responses.COULD_NOT_VALIDATE_CRED,
- headers={"WWW-Authenticate": "Bearer"},
- )
- try:
- payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
- id: str = payload.get("id")
- if id is None:
- raise credentials_exception
- token_data = id
- except JWTError:
- raise credentials_exception
- user = cls.fetch(id=token_data,db=db)
- if user is None:
- raise credentials_exception
- return user
-
- @classmethod
- def authenticate_user(cls, db: Session, password: str,email: str) -> user_schema.User:
- user = cls.fetch_by_email(email=email, db=db)
- if not user:
- return False
- if not cls.verify_password(password, user.password):
- return False
- return user
-
- @staticmethod
- def verify_password(password, hashed_password):
- return pwd_context.verify(password, hashed_password)
-
- @staticmethod
- def hash_password(password) -> str:
- return pwd_context.hash(password)
-
- @staticmethod
- def create_access_token(data: dict, db: Session, expires_delta: timedelta = None) -> str:
- to_encode = data.copy()
- if expires_delta:
- expire = datetime.utcnow() + expires_delta
- else:
- expire = datetime.utcnow() + timedelta(minutes=30)
- to_encode.update({"exp": expire})
- encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
-
- db.commit()
-
- return encoded_jwt
-
- @staticmethod
- def create_refresh_token(data: dict, db: Session) -> str:
- to_encode = data.copy()
-
- expire = datetime.utcnow() + timedelta(seconds=int(JWT_REFRESH_EXPIRY))
- to_encode.update({"exp": expire})
- encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
-
- return encoded_jwt
-
- @classmethod
- def verify_access_token(cls, token: str, db: Session) -> user_schema.TokenData:
- try:
- invalid_token = cls.check_token_blacklist(db=db, token=token)
- if invalid_token == True:
- raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail=responses.INVALID_CREDENTIALS,
- headers={"WWW-Authenticate": "Bearer"})
-
- payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
- id: int = payload.get("id")
-
- if id is None:
- raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail=responses.INVALID_CREDENTIALS,
- headers={"WWW-Authenticate": "Bearer"})
-
- user = cls.fetch(db=db,id=id)
-
- if user is None:
- raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=responses.INVALID_CREDENTIALS,
- headers={"WWW-Authenticate": "Bearer"})
-
-
- token_data = user_schema.TokenData(email=user.email, id=id)
-
- return token_data
-
- except JWTError as error:
- print(error, 'error')
- return JWTError(HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail=responses.INVALID_CREDENTIALS,
- headers={"WWW-Authenticate": "Bearer"}))
-
- @classmethod
- def verify_refresh_token(cls, refresh_token: str, db: Session) -> user_schema.TokenData:
- try:
- if not refresh_token:
- raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=responses.EXPIRED)
-
- payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
- id: str = payload.get("id")
-
- if id is None:
- raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail=responses.INVALID_CREDENTIALS,
- headers={"WWW-Authenticate": "Bearer"})
-
- user = cls.fetch(id=id, db=db)
-
- if user is None:
- raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=responses.INVALID_CREDENTIALS)
-
-
- token_data = user_schema.TokenData(email=user.email, id=id)
-
- except JWTError:
- raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail=responses.INVALID_CREDENTIALS,
- headers={"WWW-Authenticate": "Bearer"})
-
- return token_data
-
- @staticmethod
- def check_token_blacklist(token: str, db:Session)-> bool:
- fetched_token = db.query(BlackListToken).filter(BlackListToken.token == token).first()
-
- if fetched_token:
- return True
- else:
- return False
-
- @staticmethod
- def logout(token: str, user: user_schema.ShowUser, db:Session) -> str:
- blacklist_token = BlackListToken(
- token=token.split(' ')[1],
- created_by=user.id
- )
-
- db.add(blacklist_token)
- db.commit()
-
- return token
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-