Merge pull request #8 from fga-eps-mds/admincrud

59 CRUD de Admin

requirements.txt

@@ -45,8 +45,9 @@ pydantic-settings==2.0.3
 pydantic_core==2.10.1
 pylint==3.0.2
 PyMySQL==1.1.0
-pytest==7.4.2
+pytest==7.4.3
 pytest-asyncio==0.21.1
+pytest-env==1.1.1
 pytest-html==4.1.1
 pytest-metadata==3.0.0
 pytest-mock==3.12.0
@@ -66,6 +67,7 @@ tomlkit==0.12.2
 typing_extensions==4.8.0
 urllib3==2.0.7
 uvicorn==0.23.2
+uvloop==0.18.0
 watchfiles==0.21.0
 websocket-client==1.6.4
 websockets==11.0.3

src/constants/errorMessages.py

@@ -12,4 +12,5 @@
 ERROR_SENDING_EMAIL = "Ocorreu um erro ao enviar o email."
 NO_RESET_PASSWORD_CODE = "Código de reinicialização de senha não foi informado."
 INVALID_RESET_PASSWORD_CODE = "Código de reinicialização de senha está inválido."
-INVALID_REQUEST = "Requisição inválida."
+INVALID_REQUEST = "Requisição inválida."
+NO_PERMISSION = "NO PERMISSION"

src/controller/authController.py

@@ -59,8 +59,14 @@ async def login(data: authSchema.UserLogin, db: Session = Depends(get_db)):
   if not user.is_active:
     raise HTTPException(status_code=401, detail=errorMessages.ACCOUNT_IS_NOT_ACTIVE)
 
-  access_token = security.create_access_token(data={"id": user.id, "email": user.email})
+  access_token = security.create_access_token(data={ "id": user.id, "email": user.email, "role": user.role })
+  refresh_token = security.create_refresh_token(data={ "id": user.id })
 
+  return JSONResponse(status_code=200, content={ "access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer" })
+
+@auth.post("/refresh", response_model=authSchema.RefreshTokenResponse)
+def refresh_token(token: dict = Depends(security.verify_token)):
+  access_token=security.create_access_token(token)
   return JSONResponse(status_code=200, content={ "access_token": access_token, "token_type": "bearer" })
 
 @auth.post('/resend-code')
@@ -73,9 +79,6 @@ async def send_new_code(data: authSchema.SendNewCode, db: Session = Depends(get_
     return JSONResponse(status_code=400, content={ "status": "error", "message": errorMessages.ACCOUNT_ALREADY_ACTIVE })
 
   res = await send_mail.send_verification_code(email=data.email, code=user.activation_code)
-  if res.status_code != 200:
-    return JSONResponse(status_code=400, content={ "status": "error", "message": errorMessages.ERROR_SENDING_EMAIL })
-
   return JSONResponse(status_code=201, content={ "status": "success" })
 
 @auth.patch('/activate-account')

src/controller/userController.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, HTTPException, Response, status, Depends
+from fastapi import APIRouter, HTTPException, Response, status, Depends, Header
 from database import get_db
 from sqlalchemy.orm import Session
 
@@ -7,14 +7,25 @@
 from repository import userRepository
 from utils import security, enumeration
 
+from fastapi_filter import FilterDepends
 
 user = APIRouter(
   prefix="/users"
 )
 
 @user.get("/", response_model=list[userSchema.User])
-async def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db), token: dict = Depends(security.verify_token)):
-  users = userRepository.get_users(db, skip=skip, limit=limit)
+def read_users(
+  response: Response,
+  users_filter: userSchema.UserListFilter = FilterDepends(userSchema.UserListFilter),
+  db: Session = Depends(get_db), 
+  _: dict = Depends(security.verify_token),
+):
+  result = userRepository.get_users(db, users_filter)
+
+  users = result['users']
+  total = result['total']
+
+  response.headers['X-Total-Count'] = str(total)
   return users
 
 @user.get("/{user_id}", response_model=userSchema.User)
@@ -41,6 +52,11 @@ async def partial_update_user(user_id: int, data: userSchema.UserUpdate, db: Ses
   if not db_user:
     raise HTTPException(status_code=404, detail=errorMessages.USER_NOT_FOUND)
 
+  if data.email and data.email != db_user.email:
+    user = userRepository.get_user_by_email(db, data.email)
+    if user: 
+      raise HTTPException(status_code=404, detail=errorMessages.EMAIL_ALREADY_REGISTERED)
+
   updated_user = userRepository.update_user(db, db_user, data)
   return updated_user
 
@@ -52,3 +68,20 @@ async def delete_user(user_id: int, db: Session = Depends(get_db), token: dict =
 
   userRepository.delete_user(db, db_user)
   return db_user
+
+@user.patch("/role/{user_id}", response_model=userSchema.User)
+def update_role(user_id: int, db: Session = Depends(get_db), token: dict = Depends(security.verify_token)):
+  user = userRepository.get_user_by_email(db, email=token['email'])
+  if user.role != enumeration.UserRole.ADMIN.value:
+    raise HTTPException(status_code=401, detail=errorMessages.NO_PERMISSION)
+
+  # Verificar se o usuario existe
+  user = userRepository.get_user(db, user_id)
+
+  if not user:
+    raise HTTPException(status_code=404, detail=errorMessages.USER_NOT_FOUND)
+
+  new_role = enumeration.UserRole.ADMIN.value if user.role == enumeration.UserRole.USER.value else enumeration.UserRole.USER.value
+  user = userRepository.update_user_role(db, db_user=user, role=new_role)
+
+  return user

src/database.py

@@ -6,16 +6,19 @@
 POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD")
 POSTGRES_HOST = os.getenv("POSTGRES_HOST")
 POSTGRES_DB = os.getenv("POSTGRES_DB")
+POSTGRES_PORT = os.getenv("POSTGRES_PORT", default=5432)
 
-engine = create_engine(f'postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}/{POSTGRES_DB}')
+POSTGRES_URI = f'postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}'
+
+engine = create_engine(POSTGRES_URI)
 
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
 Base = declarative_base()
 
 def get_db():
-    db = SessionLocal()
-    try:
-        yield db
-    finally:
-        db.close()
+  db = SessionLocal()
+  try:
+      yield db
+  finally:
+      db.close()

src/domain/authSchema.py

@@ -12,6 +12,11 @@ class UserCreate(BaseModel):
   password: str
 
 class Token(BaseModel):
+  access_token: str
+  refresh_token: str
+  token_type: str
+
+class RefreshTokenResponse(BaseModel):
   access_token: str
   token_type: str
 

src/domain/userSchema.py

@@ -1,5 +1,9 @@
+from typing import Optional
 from pydantic import BaseModel, ConfigDict
-from typing import Union
+from fastapi_filter import FilterDepends, with_prefix
+from sqlalchemy import or_
+from fastapi_filter.contrib.sqlalchemy import Filter
+from model import userModel
 
 class UserUpdate(BaseModel):
   name: str | None = None
@@ -14,3 +18,17 @@ class User(BaseModel):
   email: str
   role: str
   is_active: bool
+
+class UserListFilter(Filter):
+  name: Optional[str] = None
+  name__like: Optional[str] = None
+  email: Optional[str] = None
+  email__like: Optional[str] = None
+  connection: Optional[str] = None
+  name_or_email: Optional[str] = None
+  offset: Optional[int] = 0
+  limit: Optional[int] = 100
+
+  class Constants(Filter.Constants):
+    model = userModel.User
+    search_model_fields = ["name", "email"]

src/main.py

@@ -24,6 +24,7 @@
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
+    expose_headers=["*"]
 )
 
 # Routers
@@ -36,7 +37,7 @@
 def read_root():
     return {"message": "UnB-TV!"}
 
-if __name__ == '__main__':
+if __name__ == '__main__': # pragma: no cover
   port = 8000
   if (len(sys.argv) == 2):
     port = sys.argv[1]

src/repository/userRepository.py

@@ -1,6 +1,7 @@
 from repository import userRepository
 
 # Referencia: https://fastapi.tiangolo.com/tutorial/sql-databases/#crud-utils
+from sqlalchemy import or_
 from sqlalchemy.orm import Session
 
 from domain import userSchema
@@ -12,8 +13,29 @@ def get_user(db: Session, user_id: int):
 def get_user_by_email(db: Session, email: str):
   return db.query(userModel.User).filter(userModel.User.email == email).first()
 
-def get_users(db: Session, skip: int = 0, limit: int = 100):
-  return db.query(userModel.User).offset(skip).limit(limit).all()
+def get_users(db: Session, users_filter: userSchema.UserListFilter):
+  query = db.query(userModel.User)
+
+  if (users_filter.name):
+    query = query.filter(userModel.User.name == users_filter.name)
+  elif (users_filter.email):
+    query = query.filter(userModel.User.email == users_filter.email)
+  elif (users_filter.name_or_email):
+    query = query.filter(or_(userModel.User.name.ilike(f'%{users_filter.name_or_email}%'), userModel.User.email.ilike(f'%{users_filter.name_or_email}%')))
+
+  if (users_filter.connection):
+    query = query.filter(userModel.User.connection == users_filter.connection)
+
+  total_count = query.count()
+  query = query.order_by(userModel.User.name.asc())
+
+  if (users_filter.offset):
+    query = query.offset(users_filter.offset)
+
+  if (users_filter.limit):
+    query = query.limit(users_filter.limit)
+
+  return { "users": query.all(), "total": total_count }
 
 def create_user(db: Session, name, connection, email, password, activation_code):
   db_user = userModel.User(name=name, connection=connection, email=email, password=password, activation_code=activation_code,)
@@ -53,6 +75,14 @@ def update_user(db: Session, db_user: userSchema.User, user: userSchema.UserUpda
   db.refresh(db_user)
   return db_user
 
+def update_user_role(db: Session, db_user: userSchema.User, role: str):
+  db_user.role = role
+
+  db.add(db_user)
+  db.commit()
+  db.refresh(db_user)
+  return db_user
+
 def update_password(db: Session, db_user: userSchema.User, new_password: str):
   db_user.password = new_password
   db_user.password_reset_code = None

src/utils/dotenv.py

@@ -2,7 +2,7 @@
 from constants import errorMessages
 
 def validate_dotenv():
-  required_env_var = ["SECRET", "ALGORITHM", "ACCESS_TOKEN_EXPIRE_MINUTES", "MAIL_USERNAME", "MAIL_PASSWORD", "MAIL_FROM", "MAIL_PORT", "MAIL_SERVER"]
+  required_env_var = ["SECRET", "ALGORITHM", "MAIL_USERNAME", "MAIL_PASSWORD", "MAIL_FROM", "MAIL_PORT", "MAIL_SERVER"]
   missing_env_var = [var for var in required_env_var if var not in os.environ]
 
   if missing_env_var:

src/utils/enumeration.py

@@ -10,4 +10,8 @@ class UserConnection(Enum):
 
   @classmethod
   def has_value(cls, value):
-    return value in cls._value2member_map_
+    return value in cls._value2member_map_
+
+class UserRole(Enum):
+  ADMIN = "ADMIN"
+  USER = "USER"

src/utils/security.py

@@ -8,7 +8,8 @@
 
 SECRET_KEY = os.getenv("SECRET")
 ALGORITHM = os.getenv("ALGORITHM")
-ACCESS_TOKEN_EXPIRE_MINUTES = os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", default=15)
+ACCESS_TOKEN_EXPIRE_MINUTES = os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", default=30)
+REFRESH_TOKEN_EXPIRE_DAYS = os.getenv("REFRESH_TOKEN_EXPIRE_DAYS", default=7)
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
@@ -29,7 +30,7 @@ def create_access_token(data: dict):
   to_encode = data.copy()
   expire = datetime.utcnow() + access_token_expires
 
-  to_encode.update({"exp": expire})
+  to_encode.update({ "exp": expire, **data })
   encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
   return encoded_jwt
 
@@ -41,4 +42,15 @@ def verify_token(token: str = Depends(oauth2_scheme)):
     raise HTTPException(status_code=401, detail=errorMessages.INVALID_TOKEN)
 
 def generate_six_digit_number_code():
-  return secrets.randbelow(900000) + 100000
+  return secrets.randbelow(900000) + 100000
+
+def create_refresh_token(data:dict):
+  access_token_expires = timedelta(days=int(REFRESH_TOKEN_EXPIRE_DAYS))
+
+  to_encode = data.copy()
+  if access_token_expires:
+    expire = datetime.utcnow() + access_token_expires
+
+  to_encode.update({"exp": expire})
+  encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+  return encoded_jwt