Add users

src/services/users.py

@@ -0,0 +1,102 @@
+from email.headerregistry import Address
+import abc
+from dataclasses import dataclass
+from src.utils.secret import SecretBox
+from src.config.email import EmailAddress
+
+
+@dataclass
+class UserProfile:
+    id: int
+    email: EmailAddress
+
+
+@dataclass
+class User:
+    id: int
+    email: EmailAddress
+    email_confirmed: bool
+    password_hash: SecretBox[str]
+
+    def to_profile(self) -> UserProfile:
+        return UserProfile(id=self.id, email=self.email)
+
+
+@dataclass
+class UserDTO:
+    email: EmailAddress
+    raw_password: SecretBox[str]
+
+
+class LoginError(Exception):
+    pass
+
+
+class SignupError(Exception):
+    pass
+
+
+class UserRepo(abc.ABC):
+    @abc.abstractmethod
+    def create_user(self, user: UserDTO) -> User:
+        pass
+
+    @abc.abstractmethod
+    def get_user_by_id(self, user_id: int) -> User | None:
+        pass
+
+    @abc.abstractmethod
+    def get_user_by_email(self, email: EmailAddress) -> User | None:
+        pass
+
+    @abc.abstractmethod
+    def update_user(self, user: User):
+        pass
+
+    @abc.abstractmethod
+    def auth_as_user(self, user: UserDTO) -> User | None:
+        pass
+
+
+def is_valid_email(email: str) -> bool:
+    try:
+        parsed = Address(addr_spec=email)
+        return bool(parsed.username and parsed.domain and '.' in parsed.domain)
+    except (ValueError, TypeError):
+        return False
+
+
+def is_valid_password(password: str) -> bool:
+    # TODO: Enforce other or saner rules?
+    long_enough = len(password) > 8
+    short_enough = len(password) < 32
+    has_symbol = any([s in password for s in list('@#$%^&*!?/')])
+    return long_enough and short_enough and has_symbol
+
+
+class UserService:
+    def __init__(self, repo: UserRepo):
+        self._repo = repo
+
+    def login(self, user: UserDTO) -> UserProfile | None:
+        full_user = self._repo.auth_as_user(user)
+        if full_user:
+            return full_user.to_profile()
+
+    def signup(self, user: UserDTO) -> UserProfile:
+        if not is_valid_email(user.email):
+            raise SignupError(f'{user.email} was not an acceptable email address')
+
+        if not is_valid_password(user.raw_password.expose_secret()):
+            raise SignupError('The given password was not acceptable')
+
+        return self._repo.create_user(user).to_profile()
+
+    def get_user_by_id(self, user_id: int) -> UserProfile | None:
+        user = self._repo.get_user_by_id(user_id)
+        if user:
+            return user.to_profile()
+
+
+def get_service(repo: UserRepo) -> UserService:
+    return UserService(repo)