refactor: moved json user management to a separate repository

selfprivacy_api/actions/users.py

@@ -2,8 +2,6 @@
 
 import re
 from typing import Optional
-from pydantic import BaseModel
-from enum import Enum
 from selfprivacy_api.utils import (
     ReadUserData,
     WriteUserData,
@@ -11,21 +9,21 @@ from selfprivacy_api.utils import (
     is_username_forbidden,
 )
 
+from selfprivacy_api.repositories.users.abstract_user_repository import (
+    UserDataUser,
+    UserDataUserOrigin,
+)
 
-class UserDataUserOrigin(Enum):
+from selfprivacy_api.repositories.users.exceptions import (
-    """Origin of the user in the user data"""
+    InvalidConfiguration,
-
+    PasswordIsEmpty,
-    NORMAL = "NORMAL"
+    UserAlreadyExists,
-    PRIMARY = "PRIMARY"
+    UserIsProtected,
-    ROOT = "ROOT"
+    UsernameForbidden,
-
+    UsernameNotAlphanumeric,
-
+    UsernameTooLong,
-class UserDataUser(BaseModel):
+    UserNotFound,
-    """The user model from the userdata file"""
+)
-
-    username: str
-    ssh_keys: list[str]
-    origin: UserDataUserOrigin
 
 
 def ensure_ssh_and_users_fields_exist(data):
@@ -78,43 +76,7 @@ def get_users(
     return users
 
 
-class UsernameForbidden(Exception):
+def create_user(username: str, password: str) -> None:
-    """Attemted to create a user with a forbidden username"""
-
-    pass
-
-
-class UserAlreadyExists(Exception):
-    """Attemted to create a user that already exists"""
-
-    pass
-
-
-class UsernameNotAlphanumeric(Exception):
-    """Attemted to create a user with a non-alphanumeric username"""
-
-    pass
-
-
-class UsernameTooLong(Exception):
-    """Attemted to create a user with a too long username. Username must be less than 32 characters"""
-
-    pass
-
-
-class PasswordIsEmpty(Exception):
-    """Attemted to create a user with an empty password"""
-
-    pass
-
-
-class InvalidConfiguration(Exception):
-    """The userdata is broken"""
-
-    pass
-
-
-def create_user(username: str, password: str):
     if password == "":
         raise PasswordIsEmpty("Password is empty")
 
@@ -150,19 +112,7 @@ def create_user(username: str, password: str):
         )
 
 
-class UserNotFound(Exception):
+def delete_user(username: str) -> None:
-    """Attemted to get a user that does not exist"""
-
-    pass
-
-
-class UserIsProtected(Exception):
-    """Attemted to delete a user that is protected"""
-
-    pass
-
-
-def delete_user(username: str):
     with WriteUserData() as user_data:
         ensure_ssh_and_users_fields_exist(user_data)
         if username == user_data["username"] or username == "root":
@@ -176,7 +126,7 @@ def delete_user(username: str):
             raise UserNotFound("User did not exist")
 
 
-def update_user(username: str, password: str):
+def update_user(username: str, password: str) -> None:
     if password == "":
         raise PasswordIsEmpty("Password is empty")
 

selfprivacy_api/graphql/common_types/user.py

@@ -1,7 +1,7 @@
 import typing
 from enum import Enum
 import strawberry
-import selfprivacy_api.actions.users as users_actions
+from selfprivacy_api.repositories.users import ACTIVE_USERS_PROVIDER as users_actions
 
 from selfprivacy_api.graphql.mutations.mutation_interface import (
     MutationReturnInterface,
@@ -31,7 +31,7 @@ class UserMutationReturn(MutationReturnInterface):
 
 
 def get_user_by_username(username: str) -> typing.Optional[User]:
-    user = users_actions.get_user_by_username(username)
+    user = users_actions.get_user_by_username(username=username)
     if user is None:
         return None
 

selfprivacy_api/graphql/mutations/deprecated_mutations.py

@@ -13,7 +13,6 @@ from selfprivacy_api.graphql.mutations.api_mutations import (
     ApiMutations,
     DeviceApiTokenMutationReturn,
 )
-from selfprivacy_api.graphql.mutations.backup_mutations import BackupMutations
 from selfprivacy_api.graphql.mutations.job_mutations import JobMutations
 from selfprivacy_api.graphql.mutations.mutation_interface import (
     GenericJobMutationReturn,
@@ -30,7 +29,6 @@ from selfprivacy_api.graphql.mutations.system_mutations import (
     SystemMutations,
     TimezoneMutationReturn,
 )
-from selfprivacy_api.graphql.mutations.backup_mutations import BackupMutations
 from selfprivacy_api.graphql.mutations.users_mutations import UsersMutations
 
 

selfprivacy_api/graphql/mutations/users_mutations.py

@@ -2,8 +2,8 @@
 """Users management module"""
 # pylint: disable=too-few-public-methods
 import strawberry
+
 from selfprivacy_api.graphql import IsAuthenticated
-from selfprivacy_api.actions.users import UserNotFound
 from selfprivacy_api.graphql.common_types.user import (
     UserMutationReturn,
     get_user_by_username,
@@ -18,7 +18,17 @@ from selfprivacy_api.actions.ssh import (
 from selfprivacy_api.graphql.mutations.mutation_interface import (
     GenericMutationReturn,
 )
-import selfprivacy_api.actions.users as users_actions
+from selfprivacy_api.repositories.users import ACTIVE_USERS_PROVIDER as users_actions
+from selfprivacy_api.repositories.users.exceptions import (
+    PasswordIsEmpty,
+    UsernameForbidden,
+    InvalidConfiguration,
+    UserAlreadyExists,
+    UserIsProtected,
+    UsernameNotAlphanumeric,
+    UsernameTooLong,
+    UserNotFound,
+)
 
 
 @strawberry.input
@@ -45,37 +55,37 @@ class UsersMutations:
     def create_user(self, user: UserMutationInput) -> UserMutationReturn:
         try:
             users_actions.create_user(user.username, user.password)
-        except users_actions.PasswordIsEmpty as e:
+        except PasswordIsEmpty as e:
             return UserMutationReturn(
                 success=False,
                 message=str(e),
                 code=400,
             )
-        except users_actions.UsernameForbidden as e:
+        except UsernameForbidden as e:
             return UserMutationReturn(
                 success=False,
                 message=str(e),
                 code=409,
             )
-        except users_actions.UsernameNotAlphanumeric as e:
+        except UsernameNotAlphanumeric as e:
             return UserMutationReturn(
                 success=False,
                 message=str(e),
                 code=400,
             )
-        except users_actions.UsernameTooLong as e:
+        except UsernameTooLong as e:
             return UserMutationReturn(
                 success=False,
                 message=str(e),
                 code=400,
             )
-        except users_actions.InvalidConfiguration as e:
+        except InvalidConfiguration as e:
             return UserMutationReturn(
                 success=False,
                 message=str(e),
                 code=400,
             )
-        except users_actions.UserAlreadyExists as e:
+        except UserAlreadyExists as e:
             return UserMutationReturn(
                 success=False,
                 message=str(e),
@@ -94,13 +104,13 @@ class UsersMutations:
     def delete_user(self, username: str) -> GenericMutationReturn:
         try:
             users_actions.delete_user(username)
-        except users_actions.UserNotFound as e:
+        except UserNotFound as e:
             return GenericMutationReturn(
                 success=False,
                 message=str(e),
                 code=404,
             )
-        except users_actions.UserIsProtected as e:
+        except UserIsProtected as e:
             return GenericMutationReturn(
                 success=False,
                 message=str(e),
@@ -118,13 +128,13 @@ class UsersMutations:
         """Update user mutation"""
         try:
             users_actions.update_user(user.username, user.password)
-        except users_actions.PasswordIsEmpty as e:
+        except PasswordIsEmpty as e:
             return UserMutationReturn(
                 success=False,
                 message=str(e),
                 code=400,
             )
-        except users_actions.UserNotFound as e:
+        except UserNotFound as e:
             return UserMutationReturn(
                 success=False,
                 message=str(e),

selfprivacy_api/repositories/users/__init__.py

@@ -0,0 +1,3 @@
+from selfprivacy_api.repositories.users.json_user_repository import JsonUserRepository
+
+ACTIVE_USERS_PROVIDER = JsonUserRepository

selfprivacy_api/repositories/users/abstract_user_repository.py

@@ -0,0 +1,47 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+
+from pydantic import BaseModel
+from enum import Enum
+
+
+class UserDataUserOrigin(Enum):
+    """Origin of the user in the user data"""
+
+    NORMAL = "NORMAL"
+    PRIMARY = "PRIMARY"
+    ROOT = "ROOT"
+
+
+class UserDataUser(BaseModel):
+    """The user model from the userdata file"""
+
+    username: str
+    ssh_keys: list[str]
+    origin: UserDataUserOrigin
+
+
+class AbstractUserRepository(ABC):
+
+    @abstractmethod
+    def get_users(
+        exclude_primary: bool = False,
+        exclude_root: bool = False,
+    ) -> list[UserDataUser]:
+        """Retrieves a list of users with options to exclude specific user groups"""
+
+    @abstractmethod
+    def create_user(username: str, password: str):
+        """Creates a new user"""
+
+    @abstractmethod
+    def delete_user(username: str) -> None:
+        """Deletes an existing user"""
+
+    @abstractmethod
+    def update_user(username: str, password: str) -> None:
+        """Updates the password of an existing user"""
+
+    @abstractmethod
+    def get_user_by_username(username: str) -> Optional[UserDataUser]:
+        """Retrieves user data (UserDataUser) by username"""

selfprivacy_api/repositories/users/exceptions.py

@@ -0,0 +1,30 @@
+class UserNotFound(Exception):
+    """Attemted to get a user that does not exist"""
+
+
+class UserIsProtected(Exception):
+    """Attemted to delete a user that is protected"""
+
+
+class UsernameForbidden(Exception):
+    """Attemted to create a user with a forbidden username"""
+
+
+class UserAlreadyExists(Exception):
+    """Attemted to create a user that already exists"""
+
+
+class UsernameNotAlphanumeric(Exception):
+    """Attemted to create a user with a non-alphanumeric username"""
+
+
+class UsernameTooLong(Exception):
+    """Attemted to create a user with a too long username. Username must be less than 32 characters"""
+
+
+class PasswordIsEmpty(Exception):
+    """Attemted to create a user with an empty password"""
+
+
+class InvalidConfiguration(Exception):
+    """The userdata is broken"""

selfprivacy_api/repositories/users/json_user_repository.py

@@ -0,0 +1,38 @@
+from typing import Optional
+
+from selfprivacy_api.repositories.users.abstract_user_repository import (
+    AbstractUserRepository,
+    UserDataUser,
+)
+
+from selfprivacy_api.actions.users import (
+    create_user,
+    delete_user,
+    get_user_by_username,
+    get_users,
+    update_user,
+)
+
+
+class JsonUserRepository(AbstractUserRepository):
+    def get_users(
+        exclude_primary: bool = False,
+        exclude_root: bool = False,
+    ) -> list[UserDataUser]:
+        return get_users(exclude_primary=exclude_primary, exclude_root=exclude_root)
+
+    def create_user(username: str, password: str):
+        """Creates a new user"""
+        return create_user(username=username, password=password)
+
+    def delete_user(username: str) -> None:
+        """Deletes an existing user"""
+        return delete_user(username=username)
+
+    def update_user(username: str, password: str) -> None:
+        """Updates the password of an existing user"""
+        return update_user(username=username, password=password)
+
+    def get_user_by_username(username: str) -> Optional[UserDataUser]:
+        """Retrieves user data (UserDataUser) by username"""
+        return get_user_by_username(username=username)