Customize directory structure

ara/controller/authentication.py

@@ -0,0 +1,3 @@
+class AuthLoggedInUser:
+    # TODO
+    pass

ara/controller/constants.py

@@ -0,0 +1,65 @@
+from enum import IntEnum, unique
+
+
+@unique
+class HttpStatusCode(IntEnum):
+    CONTINUE = 100
+    SWITCHING_PROTOCOLS = 101
+    OK = 200
+    CREATED = 201
+    ACCEPTED = 202
+    NON_AUTHORITATIVE_INFORMATION = 203
+    NO_CONTENT = 204
+    RESET_CONTENT = 205
+    PARTIAL_CONTENT = 206
+    MULTI_STATUS = 207
+    ALREADY_REPORTED = 208
+    IM_USED = 226
+    MULTIPLE_CHOICES = 300
+    MOVED_PERMANENTLY = 301
+    FOUND = 302
+    SEE_OTHER = 303
+    NOT_MODIFIED = 304
+    USE_PROXY = 305
+    RESERVED = 306
+    TEMPORARY_REDIRECT = 307
+    PERMANENT_REDIRECT = 308
+    BAD_REQUEST = 400
+    UNAUTHORIZED = 401
+    PAYMENT_REQUIRED = 402
+    FORBIDDEN = 403
+    NOT_FOUND = 404
+    METHOD_NOT_ALLOWED = 405
+    NOT_ACCEPTABLE = 406
+    PROXY_AUTHENTICATION_REQUIRED = 407
+    REQUEST_TIMEOUT = 408
+    CONFLICT = 409
+    GONE = 410
+    LENGTH_REQUIRED = 411
+    PRECONDITION_FAILED = 412
+    REQUEST_ENTITY_TOO_LARGE = 413
+    REQUEST_URI_TOO_LONG = 414
+    UNSUPPORTED_MEDIA_TYPE = 415
+    REQUESTED_RANGE_NOT_SATISFIABLE = 416
+    EXPECTATION_FAILED = 417
+    IM_A_TEAPOT = 418
+    UNPROCESSABLE_ENTITY = 422
+    LOCKED = 423
+    FAILED_DEPENDENCY = 424
+    UPGRADE_REQUIRED = 426
+    PRECONDITION_REQUIRED = 428
+    TOO_MANY_REQUESTS = 429
+    REQUEST_HEADER_FIELDS_TOO_LARGE = 431
+    UNAVAILABLE_FOR_LEGAL_REASONS = 451
+    INTERNAL_SERVER_ERROR = 500
+    NOT_IMPLEMENTED = 501
+    BAD_GATEWAY = 502
+    SERVICE_UNAVAILABLE = 503
+    GATEWAY_TIMEOUT = 504
+    HTTP_VERSION_NOT_SUPPORTED = 505
+    VARIANT_ALSO_NEGOTIATES = 506
+    INSUFFICIENT_STORAGE = 507
+    LOOP_DETECTED = 508
+    BANDWIDTH_LIMIT_EXCEEDED = 509
+    NOT_EXTENDED = 510
+    NETWORK_AUTHENTICATION_REQUIRED = 511

ara/controller/pagination.py

@@ -0,0 +1,14 @@
+from typing import Any, Optional
+
+from pydantic import BaseModel
+
+
+class PaginatedData(BaseModel):
+    count: int
+    next: Optional[str]
+    previous: Optional[str]
+    results: list[Any]
+
+
+class NesAraPagination:
+    pass

ara/controller/request.py

@@ -0,0 +1,10 @@
+from typing import TypeVar
+
+from django.contrib.auth import get_user_model
+from django.http import HttpRequest
+
+User = TypeVar("User", bound=get_user_model())
+
+
+class LoggedInUserRequest(HttpRequest):
+    user: User

ara/controller/response.py

@@ -0,0 +1,21 @@
+from http import HTTPStatus
+from typing import Any, NamedTuple, Optional
+
+from pydantic import BaseModel
+
+
+class NewAraResponse(NamedTuple):
+    status_code: HTTPStatus
+    data: Any
+
+
+class NewAraErrorResponseBody(BaseModel):
+    # necessary
+    error_code: Optional[int]
+    error_reason: str = ""
+
+    def __init__(self, exception: Exception):
+        message = str(exception)
+        # TODO: Create NewAraException and use error_code & error_reason
+        data = {"message": message}
+        super().__init__(**data)

ara/domain/ara_entity.py

@@ -0,0 +1,35 @@
+from typing import Any
+
+from pydantic import BaseModel, PrivateAttr
+
+
+class NewAraEntity(BaseModel):
+    _updated_fields: set[str] = PrivateAttr(set())
+
+    @property
+    def updated_fields(self):
+        return self._updated_fields
+
+    @property
+    def updated_values(self) -> dict[str, Any]:
+        return {key: getattr(self, key) for key in self._updated_fields}
+
+    def set_attribute(self, field_name: str, value: Any):
+        self.__dict__[field_name] = value
+        is_private_field = field_name.startswith("_") or field_name.startswith("__")
+        if is_private_field is False:
+            self._updated_fields.add(field_name)
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        is_private_field = name.startswith("_") or name.startswith("_")
+        if is_private_field is False:
+            self._updated_fields.add(name)
+        return super().__setattr__(name, value)
+
+    class Config:
+        arbitrary_types_allowed = True
+
+
+class NewAraEntityCreateInput(BaseModel):
+    class Config:
+        arbitrary_types_allowed = True

ara/infra/django_infra.py

@@ -0,0 +1,152 @@
+from typing import Any, Generic, Optional, Type, TypeVar, Union
+
+from django.core.exceptions import ObjectDoesNotExist, ValidationError
+from django.db.models import Model
+
+from ara.domain.ara_entity import NewAraEntity, NewAraEntityCreateInput
+
+T = TypeVar("T", bound=Model)
+
+
+class NewAraDjangoInfra(Generic[T]):
+    def __init__(self, model: Type[T]) -> None:
+        self.model = model
+
+    def get_by_id(self, id: Any, *, is_select_for_update: bool = False) -> T:
+        """
+        Generic function for simple get by id queries.
+
+        Args:
+            id (Any):
+                Not sure of the id type. It could be hash Id or int.
+                TODO(hyuk): check for the all models.
+            is_select_for_update (bool):
+                Set True if get queryset for update purpose. Defaults to False.
+
+        """
+        if is_select_for_update:
+            return self.model.objects.select_for_update().get(id=id)
+        return self.model.objects.get(id=id)
+
+    def get_filtered_objects(
+        self,
+        *,
+        columns: Optional[list[str]] = None,
+        conditions: dict[str, Any],
+        is_select_for_update: bool = False,
+    ) -> list[Union[T, dict[str, Any]]]:
+        """
+        Generic function for simple queries.
+        Should not be used for complex & specific purpose queries.
+
+        Args:
+            columns (Optional[List[str]]):
+                List of column names to fetch. Get all columns if None. Default None.
+            conditions (Dict[str, Any]):
+                Dictionary of field names and their corresponding values to filter by.
+            is_select_for_update (bool):
+                Set True if get queryset for update purpose. Defaults to False.
+
+        Returns:
+            List[Union[T, Dict[str, Any]]]:
+                A list containing the matching object,
+                with only the specified columns if `columns` is not None.
+
+        Example:
+            # Get all rows with id=1 and only fetch 'id' and 'name' fields
+            query1 = get_filtered_queryset(
+                columns=['id', 'name'],
+                conditions={'id': 1}
+            )
+
+            # Get the first 10 rows with rating>=4.0 and order by created_at descending
+            query2 = get_filtered_queryset(
+                columns=['id', 'name', 'rating', 'created_at'],
+                conditions={'rating__gte': 4.0}
+            ).order_by('-created_at').limit(10)
+
+        Raises:
+            ValidationError: If conditions parameter is empty or invalid.
+        """
+        if not conditions:
+            raise ValidationError("conditions parameter is required")
+
+        try:
+            if is_select_for_update:
+                queryset = self.model.objects.select_for_update()
+            else:
+                queryset = self.model.objects
+
+            queryset = queryset.filter(**conditions)
+
+            if columns is not None:
+                queryset = queryset.values(*columns)
+
+            return list(queryset)
+
+        except ValidationError:
+            raise ValidationError("invalid conditions parameter")
+
+    def create_manual(self, **kwargs) -> T:
+        return self.model.objects.create(**kwargs)
+
+    def update_or_create(self, **kwargs) -> tuple[T, bool]:
+        return self.model.objects.update_or_create(**kwargs)
+
+    def get_by(self, **kwargs) -> Optional[T]:
+        """Returns repository model instance if exists.
+
+        :param kwargs: keyword arguments of fields
+        :raises MultipleObjectsXxx: when multiple rows exist
+        :return: None or model instance if exists.
+        """
+        try:
+            return self.model.objects.get(**kwargs)
+        except ObjectDoesNotExist:
+            return None
+
+    def _to_model(self, entity: NewAraEntity) -> Model:
+        raise NotImplementedError()
+
+    @staticmethod
+    def convert_model_to_entity(model: T) -> NewAraEntity:
+        raise NotImplementedError()
+
+    def _convert_entity_to_model(self, entity: NewAraEntity) -> Model:
+        raise NotImplementedError()
+
+    def _convert_create_input_to_model(
+        self, create_input: NewAraEntityCreateInput
+    ) -> T:
+        raise NotImplementedError()
+
+    def bulk_update_entity(self, entities: list[NewAraEntity]):
+        if len(entities) == 0:
+            return
+
+        model_instances = [self._convert_entity_to_model(entity) for entity in entities]
+
+        unique_updated_fields = list(
+            {field for entity in entities for field in entity.updated_fields}
+        )
+        if len(unique_updated_fields) == 0:
+            return
+
+        self.model.objects.bulk_update(model_instances, unique_updated_fields)
+
+    def bulk_update(self, instances: list[T], fields: list[str]):
+        return self.model.objects.bulk_update(instances, fields)
+
+    def bulk_create(self, inputs: list[NewAraEntityCreateInput]) -> list[NewAraEntity]:
+        instances = [self._convert_create_input_to_model(input) for input in inputs]
+        created_instances = self.model.objects.bulk_create(instances)
+        entities = [
+            self.convert_model_to_entity(created_instance)
+            for created_instance in created_instances
+        ]
+        return entities
+
+    def save_entity(self, entity: NewAraEntity):
+        model = self._convert_entity_to_model(entity)
+        model.save()
+        return entity