[feat] add basic domain

domain/error.py

@@ -0,0 +1,18 @@
+class DomainError(Exception):
+    pass
+
+
+class UserError(DomainError):
+    pass
+
+
+class UserNotFoundError(UserError):
+    def __init__(self, user_id: str) -> None:
+        super().__init__('user_not_found')
+        self.user_id = user_id
+
+
+class UserConflictError(UserError):
+    def __init__(self, email: str) -> None:
+        super().__init__('user_conflict')
+        self.email = email

domain/user.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class User:
+    id: str
+    email: str
+    name: str

repository/user.py

@@ -0,0 +1,21 @@
+from collections.abc import Iterable
+
+from domain.user import User
+from usecase.interface import UserRepository
+
+
+class InMemoryUserRepository(UserRepository):
+    def __init__(self, users: Iterable[User] | None = None) -> None:
+        self._users = {user.id: user for user in users or ()}
+
+    def get(self, user_id: str) -> User | None:
+        return self._users.get(user_id)
+
+    def get_by_email(self, email: str) -> User | None:
+        for user in self._users.values():
+            if user.email == email:
+                return user
+        return None
+
+    def save(self, user: User) -> None:
+        self._users[user.id] = user

usecase/interface.py

@@ -0,0 +1,67 @@
+from collections.abc import Mapping
+from types import TracebackType
+from typing import Protocol, TypeAlias
+
+from domain.user import User
+
+AttrValue: TypeAlias = str | int | float | bool
+Attrs: TypeAlias = Mapping[str, AttrValue]
+
+
+class UserRepository(Protocol):
+    def get(self, user_id: str) -> User | None: ...
+
+    def get_by_email(self, email: str) -> User | None: ...
+
+    def save(self, user: User) -> None: ...
+
+
+class Logger(Protocol):
+    def debug(self, message: str, attrs: Attrs | None = None) -> None: ...
+
+    def info(self, message: str, attrs: Attrs | None = None) -> None: ...
+
+    def warning(self, message: str, attrs: Attrs | None = None) -> None: ...
+
+    def error(self, message: str, attrs: Attrs | None = None) -> None: ...
+
+
+class Metrics(Protocol):
+    def increment(
+        self,
+        name: str,
+        value: int = 1,
+        attrs: Attrs | None = None,
+    ) -> None: ...
+
+    def record(
+        self,
+        name: str,
+        value: float,
+        attrs: Attrs | None = None,
+    ) -> None: ...
+
+
+class Span(Protocol):
+    def set_attribute(self, name: str, value: AttrValue) -> None: ...
+
+    def record_error(self, error: Exception) -> None: ...
+
+
+class SpanContext(Protocol):
+    def __enter__(self) -> Span: ...
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> bool | None: ...
+
+
+class Tracer(Protocol):
+    def start_span(
+        self,
+        name: str,
+        attrs: Attrs | None = None,
+    ) -> SpanContext: ...

usecase/user.py

@@ -0,0 +1,41 @@
+from dataclasses import dataclass
+
+from domain.error import UserNotFoundError
+from domain.user import User
+from usecase.interface import Logger, Tracer, UserRepository
+
+
+@dataclass(frozen=True, slots=True)
+class GetUserQuery:
+    user_id: str
+
+
+class GetUser:
+    def __init__(
+        self,
+        repository: UserRepository,
+        logger: Logger,
+        tracer: Tracer,
+    ) -> None:
+        self._repository = repository
+        self._logger = logger
+        self._tracer = tracer
+
+    def execute(self, query: GetUserQuery) -> User:
+        with self._tracer.start_span(
+            'usecase.get_user',
+            attrs={'user.id': query.user_id},
+        ) as span:
+            user = self._repository.get(query.user_id)
+            if user is None:
+                error = UserNotFoundError(query.user_id)
+                span.record_error(error)
+                self._logger.warning(
+                    'user_not_found',
+                    attrs={'user_id': query.user_id},
+                )
+                raise error
+
+            span.set_attribute('user.email', user.email)
+            self._logger.info('user_loaded', attrs={'user_id': user.id})
+            return user