refactor: database dependency injected with Database abstract class

Author: Cinnamoon-dev

src/controllers/_helpers.py

@@ -16,7 +16,7 @@ def get_current_user(token: token_dependency) -> dict[str, Any]:
     if not user_id:
         raise HTTPException(status_code=401, detail="Usuário não autorizado")
 
-    user = UserService().view(user_id)
+    user = UserService(PgDatabase()).view(user_id)
     if not user:
         raise HTTPException(status_code=404, detail="Usuário não encontrado")
 

src/controllers/authController.py

@@ -1,6 +1,7 @@
 from datetime import timedelta
 from fastapi import APIRouter, HTTPException, Header
 
+from src.infra.database.database import PgDatabase
 from src.services.userService import UserService
 from ._helpers import user_dependency, form_auth_dependency
 from src.infra.security.hashing import (
@@ -21,7 +22,7 @@
 @router.post("/login")
 def login(form_data: form_auth_dependency):
     email = form_data.username.lower()
-    user = UserService().view_by_email(email)
+    user = UserService(PgDatabase()).view_by_email(email)
 
     if not user:
         raise HTTPException(
@@ -56,7 +57,7 @@ def refresh(refresh_token: str = Header(..., alias="X-Refresh-Token")):
     payload = decode_token(refresh_token, JWT_REFRESH_SECRET_KEY, [ALGORITHM])
     user_id = int(payload["sub"])
 
-    user = UserService().view(user_id)
+    user = UserService(PgDatabase()).view(user_id)
     if not user:
         raise HTTPException(
             status_code=401, detail={"message": "User not found", "error": True}

src/controllers/userController.py

@@ -2,6 +2,7 @@
 from fastapi import APIRouter, Depends, Request
 
 from ._helpers import PermissionChecker
+from src.infra.database import PgDatabase
 from src.services.userService import UserService
 from src.schemas.userSchema import UserAddSchema, UserEditSchema
 
@@ -18,12 +19,12 @@ def user_all(
     show_fk_id: int | None = 1,
     perms=Depends(PermissionChecker("usuario-all")),
 ):
-    return UserService().all(request.query_params)
+    return UserService(PgDatabase()).all(request.query_params)
 
 
 @router.get("/{user_id:int}")
 def user_view(user_id: int, perms=Depends(PermissionChecker("usuario-view"))):
-    user = UserService().view(user_id)
+    user = UserService(PgDatabase()).view(user_id)
 
     if user is None:
         return JSONResponse(
@@ -36,7 +37,7 @@ def user_view(user_id: int, perms=Depends(PermissionChecker("usuario-view"))):
 
 @router.post("/")
 def user_add(user: UserAddSchema, perms=Depends(PermissionChecker("usuario-add"))):
-    inserted_id = UserService().add(user)
+    inserted_id = UserService(PgDatabase()).add(user)
     return JSONResponse(
         status_code=200,
         content={
@@ -51,15 +52,15 @@ def user_add(user: UserAddSchema, perms=Depends(PermissionChecker("usuario-add")
 def user_edit(
     user_id: int, user: UserEditSchema, perms=Depends(PermissionChecker("usuario-edit"))
 ):
-    UserService().edit(user_id, user)
+    UserService(PgDatabase()).edit(user_id, user)
     return JSONResponse(
         status_code=200, content={"error": False, "message": "User edited successfully"}
     )
 
 
 @router.delete("/{user_id:int}")
 def user_delete(user_id: int, perms=Depends(PermissionChecker("usuario-delete"))):
-    UserService().delete(user_id)
+    UserService(PgDatabase()).delete(user_id)
     return JSONResponse(
         status_code=200,
         content={"error": False, "message": "User deleted successfully"},

src/controllers/userTypeController.py

@@ -2,6 +2,7 @@
 from fastapi import APIRouter, Depends, Request
 
 from ._helpers import PermissionChecker
+from src.infra.database import PgDatabase
 from src.schemas.userTypeSchema import UserTypeSchema
 from src.services.userTypeService import UserTypeService
 
@@ -16,14 +17,14 @@ def user_type_all(
     sort_by: str | None = None,
     perms=Depends(PermissionChecker("tipo_usuario-all")),
 ):
-    return UserTypeService().all(request.query_params)
+    return UserTypeService(PgDatabase()).all(request.query_params)
 
 
 @router.get("/{user_type_id:int}")
 def user_type_view(
     user_type_id: int, perms=Depends(PermissionChecker("tipo_usuario-view"))
 ):
-    user_type = UserTypeService().view(user_type_id)
+    user_type = UserTypeService(PgDatabase()).view(user_type_id)
 
     if user_type is None:
         return JSONResponse(
@@ -41,7 +42,7 @@ def user_type_view(
 def user_type_add(
     user_type: UserTypeSchema, perms=Depends(PermissionChecker("tipo_usuario-add"))
 ):
-    inserted_id = UserTypeService().add(user_type)
+    inserted_id = UserTypeService(PgDatabase()).add(user_type)
     return JSONResponse(
         status_code=200,
         content={
@@ -58,7 +59,7 @@ def user_type_edit(
     user_type: UserTypeSchema,
     perms=Depends(PermissionChecker("tipo_usuario-edit")),
 ):
-    UserTypeService().edit(user_type_id, user_type)
+    UserTypeService(PgDatabase()).edit(user_type_id, user_type)
     return JSONResponse(
         status_code=200,
         content={"error": False, "message": "User type edited successfully"},
@@ -69,7 +70,7 @@ def user_type_edit(
 def user_type_delete(
     user_type_id: int, perms=Depends(PermissionChecker("tipo_usuario-delete"))
 ):
-    UserTypeService().delete(user_type_id)
+    UserTypeService(PgDatabase()).delete(user_type_id)
     return JSONResponse(
         status_code=200,
         content={"error": False, "message": "User type deleted successfully"},

src/services/userService.py

@@ -3,7 +3,7 @@
 from fastapi import HTTPException
 from fastapi.datastructures import QueryParams
 
-from src.infra.database.database import PgDatabase
+from src.infra.database.database import Database
 from src.services import paginate, fields_to_update
 from src.infra.security.hashing import bcrypt_context
 from src.infra.database import retrieve_table_columns
@@ -12,7 +12,8 @@
 
 
 class UserService:
-    def __init__(self) -> None:
+    def __init__(self, database: Database) -> None:
+        self.database = database
         self.table = "usuario"
         self.columns = retrieve_table_columns(self.table)
         try:
@@ -70,7 +71,7 @@ def view(self, user_id: int) -> dict[str, Any] | None:
         user = None
 
         try:
-            with PgDatabase() as db:
+            with self.database as db:
                 db.cursor.execute(
                     f"SELECT {self.all_columns} FROM {self.table} WHERE id = %s",
                     (user_id,),
@@ -90,7 +91,7 @@ def view_by_email(self, email: str) -> dict[str, Any] | None:
         user = None
 
         try:
-            with PgDatabase() as db:
+            with self.database as db:
                 db.cursor.execute(
                     f"SELECT {self.all_columns} FROM {self.table} WHERE email = %s",
                     (email,),
@@ -110,7 +111,7 @@ def add(self, user: UserAddSchema) -> int:
         senha = bcrypt_context.hash(user.senha)
 
         try:
-            with PgDatabase() as db:
+            with self.database as db:
                 db.cursor.execute(
                     f"INSERT INTO {self.table} (email, senha, tipo_usuario_id) VALUES (%s, %s, %s) RETURNING id",
                     (user.email.lower(), senha, user.tipo_usuario_id),
@@ -150,7 +151,7 @@ def edit(self, user_id: int, user: UserEditSchema) -> None:
 
         set_fields, set_values = fields_to_update(user_dict)
         try:
-            with PgDatabase() as db:
+            with self.database as db:
                 db.cursor.execute(
                     f"UPDATE {self.table} SET {set_fields} WHERE id = %s",
                     set_values + (user_id,),
@@ -163,7 +164,7 @@ def edit(self, user_id: int, user: UserEditSchema) -> None:
 
     def delete(self, user_id: int) -> None:
         try:
-            with PgDatabase() as db:
+            with self.database as db:
                 db.cursor.execute(f"DELETE FROM {self.table} WHERE id = %s", (user_id,))
                 db.connection.commit()
         except Exception:

src/services/userTypeService.py

@@ -4,15 +4,16 @@
 from psycopg2.errors import UniqueViolation
 from fastapi.datastructures import QueryParams
 
-from src.infra.database.database import PgDatabase
+from src.infra.database.database import Database, PgDatabase
 from src.services import paginate, fields_to_update
 from src.infra.database import retrieve_table_columns
 from src.schemas.userTypeSchema import UserTypeSchema
 from src.infra.database.serializers import line_to_dict
 
 
 class UserTypeService:
-    def __init__(self) -> None:
+    def __init__(self, database: Database) -> None:
+        self.database = database
         self.table: str = "tipo_usuario"
         self.columns: list[str] = retrieve_table_columns(self.table)
         try:
@@ -65,7 +66,7 @@ def view(self, user_type_id: int) -> dict[str, Any] | None:
         user_type = None
 
         try:
-            with PgDatabase() as db:
+            with self.database as db:
                 db.cursor.execute(
                     f"SELECT {self.all_columns} FROM {self.table} WHERE id = %s",
                     (user_type_id,),
@@ -83,7 +84,7 @@ def view(self, user_type_id: int) -> dict[str, Any] | None:
 
     def add(self, user_type: UserTypeSchema) -> int:
         try:
-            with PgDatabase() as db:
+            with self.database as db:
                 db.cursor.execute(
                     f"INSERT INTO {self.table} (nome) VALUES (%s) RETURNING id",
                     (user_type.nome,),
@@ -125,7 +126,7 @@ def edit(self, user_type_id: int, user_type: UserTypeSchema) -> None:
 
         set_fields, set_values = fields_to_update(user_type_dict)
         try:
-            with PgDatabase() as db:
+            with self.database as db:
                 db.cursor.execute(
                     f"UPDATE {self.table} SET {set_fields} WHERE id = %s",
                     set_values + (user_type_id,),
@@ -142,7 +143,7 @@ def edit(self, user_type_id: int, user_type: UserTypeSchema) -> None:
 
     def delete(self, user_type_id: int) -> None:
         try:
-            with PgDatabase() as db:
+            with self.database as db:
                 db.cursor.execute(
                     f"DELETE FROM {self.table} WHERE id = %s", (user_type_id,)
                 )