feat: create teams

Author: lakshayman

todo/dto/responses/create_team_response.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel
+from todo.dto.team_dto import TeamDTO
+
+
+class CreateTeamResponse(BaseModel):
+    team: TeamDTO
+    message: str = "Team created successfully" 

todo/dto/team_dto.py

@@ -0,0 +1,21 @@
+from pydantic import BaseModel
+from typing import List, Optional
+from datetime import datetime
+
+
+class CreateTeamDTO(BaseModel):
+    name: str
+    description: Optional[str] = None
+    member_ids: List[str] = []
+    poc_id: str
+
+
+class TeamDTO(BaseModel):
+    id: str
+    name: str
+    description: Optional[str] = None
+    poc_id: str
+    created_by: str
+    updated_by: str
+    created_at: datetime
+    updated_at: datetime

todo/models/team.py

@@ -0,0 +1,38 @@
+from pydantic import Field
+from typing import ClassVar
+from datetime import datetime, timezone
+
+from todo.models.common.document import Document
+from todo.models.common.pyobjectid import PyObjectId
+
+
+class TeamModel(Document):
+    """
+    Model for teams.
+    """
+    collection_name: ClassVar[str] = "teams"
+
+    name: str
+    description: str | None = None
+    poc_id: PyObjectId
+    created_by: PyObjectId
+    updated_by: PyObjectId
+    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    is_deleted: bool = False
+
+
+class UserTeamDetailsModel(Document):
+    """
+    Model for user-team relationships.
+    """
+    collection_name: ClassVar[str] = "userTeamDetails"
+
+    user_id: PyObjectId
+    team_id: PyObjectId
+    is_active: bool = True 
+    role_id: str
+    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    created_by: PyObjectId
+    updated_by: PyObjectId

todo/repositories/team_repository.py

@@ -0,0 +1,79 @@
+from datetime import datetime, timezone
+from typing import Optional
+from bson import ObjectId
+
+from todo.models.team import TeamModel, UserTeamDetailsModel
+from todo.repositories.common.mongo_repository import MongoRepository
+from todo.models.common.pyobjectid import PyObjectId
+
+
+class TeamRepository(MongoRepository):
+    collection_name = TeamModel.collection_name
+
+    @classmethod
+    def create(cls, team: TeamModel) -> TeamModel:
+        """
+        Creates a new team in the repository.
+        """
+        teams_collection = cls.get_collection()
+        team.created_at = datetime.now(timezone.utc)
+        team.updated_at = datetime.now(timezone.utc)
+
+        team_dict = team.model_dump(mode="json", by_alias=True, exclude_none=True)
+        insert_result = teams_collection.insert_one(team_dict)
+        team.id = insert_result.inserted_id
+        return team
+
+    @classmethod
+    def get_by_id(cls, team_id: str) -> Optional[TeamModel]:
+        """
+        Get a team by its ID.
+        """
+        teams_collection = cls.get_collection()
+        try:
+            team_data = teams_collection.find_one({"_id": ObjectId(team_id), "is_deleted": False})
+            if team_data:
+                return TeamModel(**team_data)
+            return None
+        except Exception:
+            return None
+
+
+class UserTeamDetailsRepository(MongoRepository):
+    collection_name = UserTeamDetailsModel.collection_name
+
+    @classmethod
+    def create(cls, user_team: UserTeamDetailsModel) -> UserTeamDetailsModel:
+        """
+        Creates a new user-team relationship.
+        """
+        collection = cls.get_collection()
+        user_team.created_at = datetime.now(timezone.utc)
+        user_team.updated_at = datetime.now(timezone.utc)
+        
+        user_team_dict = user_team.model_dump(mode="json", by_alias=True, exclude_none=True)
+        insert_result = collection.insert_one(user_team_dict)
+        user_team.id = insert_result.inserted_id
+        return user_team
+
+    @classmethod
+    def create_many(cls, user_teams: list[UserTeamDetailsModel]) -> list[UserTeamDetailsModel]:
+        """
+        Creates multiple user-team relationships.
+        """
+        collection = cls.get_collection()
+        current_time = datetime.now(timezone.utc)
+        
+        for user_team in user_teams:
+            user_team.created_at = current_time
+            user_team.updated_at = current_time
+        
+        user_teams_dicts = [user_team.model_dump(mode="json", by_alias=True, exclude_none=True) 
+                           for user_team in user_teams]
+        insert_result = collection.insert_many(user_teams_dicts)
+        
+        # Set the inserted IDs
+        for i, user_team in enumerate(user_teams):
+            user_team.id = insert_result.inserted_ids[i]
+        
+        return user_teams 

todo/serializers/create_team_serializer.py

@@ -0,0 +1,12 @@
+from rest_framework import serializers
+
+
+class CreateTeamSerializer(serializers.Serializer):
+    name = serializers.CharField(max_length=100)
+    description = serializers.CharField(max_length=500, required=False, allow_blank=True)
+    member_ids = serializers.ListField(
+        child=serializers.CharField(),
+        required=False,
+        default=list
+    )
+    poc_id = serializers.CharField(required=True, allow_null=False) 

todo/services/team_service.py

@@ -0,0 +1,85 @@
+from todo.dto.team_dto import CreateTeamDTO, TeamDTO
+from todo.dto.responses.create_team_response import CreateTeamResponse
+from todo.models.team import TeamModel, UserTeamDetailsModel
+from todo.models.common.pyobjectid import PyObjectId
+from todo.repositories.team_repository import TeamRepository, UserTeamDetailsRepository
+from todo.repositories.user_repository import UserRepository
+
+
+class TeamService:
+    @classmethod
+    def create_team(cls, dto: CreateTeamDTO, created_by_user_id: str) -> CreateTeamResponse:
+        """
+        Create a new team with members and POC.
+        """
+        try:
+            # Validate that all member IDs exist
+            if dto.member_ids:
+                for user_id in dto.member_ids:
+                    user = UserRepository.get_by_id(user_id)
+                    if not user:
+                        raise ValueError(f"User with ID {user_id} not found")
+
+            # Validate POC exists if provided
+            if dto.poc_id:
+                poc_user = UserRepository.get_by_id(dto.poc_id)
+                if not poc_user:
+                    raise ValueError(f"POC user with ID {dto.poc_id} not found")
+
+            # Create team
+            team = TeamModel(
+                name=dto.name,
+                description=dto.description,
+                poc_id=PyObjectId(dto.poc_id) if dto.poc_id else None,
+                created_by=PyObjectId(created_by_user_id),
+                updated_by=PyObjectId(created_by_user_id)
+            )
+
+            created_team = TeamRepository.create(team)
+
+            # Create user-team relationships
+            user_teams = []
+            
+            # Add members to the team
+            if dto.member_ids:
+                for user_id in dto.member_ids:
+                    user_team = UserTeamDetailsModel(
+                        user_id=PyObjectId(user_id),
+                        team_id=created_team.id,
+                        role_id="1",
+                        created_by=PyObjectId(created_by_user_id),
+                        updated_by=PyObjectId(created_by_user_id)
+                    )
+                    user_teams.append(user_team)
+
+            # Add creator if not already in member_ids
+            if created_by_user_id not in dto.member_ids:
+                creator_user_team = UserTeamDetailsModel(
+                    user_id=PyObjectId(created_by_user_id),
+                    team_id=created_team.id,
+                    role_id="1",
+                    created_by=PyObjectId(created_by_user_id),
+                    updated_by=PyObjectId(created_by_user_id)
+                )
+                user_teams.append(creator_user_team)
+
+            # Create all user-team relationships
+            if user_teams:
+                UserTeamDetailsRepository.create_many(user_teams)
+
+            # Convert to DTO
+            team_dto = TeamDTO(
+                id=str(created_team.id),
+                name=created_team.name,
+                description=created_team.description,
+                poc_id=str(created_team.poc_id) if created_team.poc_id else None,
+                created_by=str(created_team.created_by),
+                updated_by=str(created_team.updated_by),
+                created_at=created_team.created_at,
+                updated_at=created_team.updated_at,
+            )
+
+            return CreateTeamResponse(team=team_dto)
+
+        except Exception as e:
+            raise ValueError(str(e)) 

todo/urls.py

@@ -2,6 +2,7 @@
 from todo.views.task import TaskListView, TaskDetailView
 from todo.views.label import LabelListView
 from todo.views.health import HealthView
+from todo.views.team import TeamListView
 from todo.views.auth import (
     GoogleLoginView,
     GoogleCallbackView,
@@ -11,6 +12,7 @@
 urlpatterns = [
     path("tasks", TaskListView.as_view(), name="tasks"),
     path("tasks/<str:task_id>", TaskDetailView.as_view(), name="task_detail"),
+    path("teams", TeamListView.as_view(), name="teams"),
     path("health", HealthView.as_view(), name="health"),
     path("labels", LabelListView.as_view(), name="labels"),
     path("auth/google/login/", GoogleLoginView.as_view(), name="google_login"),

todo/views/team.py

@@ -0,0 +1,85 @@
+from rest_framework.views import APIView
+from rest_framework.response import Response
+from rest_framework import status
+from rest_framework.request import Request
+from django.conf import settings
+
+from todo.serializers.create_team_serializer import CreateTeamSerializer
+from todo.services.team_service import TeamService
+from todo.dto.team_dto import CreateTeamDTO
+from todo.dto.responses.create_team_response import CreateTeamResponse
+from todo.dto.responses.error_response import ApiErrorResponse
+from todo.constants.messages import ApiErrors
+
+
+class TeamListView(APIView):
+    def post(self, request: Request):
+        """
+        Create a new team.
+        """
+        serializer = CreateTeamSerializer(data=request.data)
+
+        if not serializer.is_valid():
+            return self._handle_validation_errors(serializer.errors)
+
+        try:
+            dto = CreateTeamDTO(**serializer.validated_data)
+            created_by_user_id = request.user_id
+            response: CreateTeamResponse = TeamService.create_team(dto, created_by_user_id)
+
+            return Response(
+                data=response.model_dump(mode="json"), 
+                status=status.HTTP_201_CREATED
+            )
+
+        except ValueError as e:
+            if isinstance(e.args[0], ApiErrorResponse):
+                error_response = e.args[0]
+                return Response(
+                    data=error_response.model_dump(mode="json"), 
+                    status=error_response.statusCode
+                )
+
+            fallback_response = ApiErrorResponse(
+                statusCode=500,
+                message=ApiErrors.UNEXPECTED_ERROR_OCCURRED,
+                errors=[{"detail": str(e) if settings.DEBUG else ApiErrors.INTERNAL_SERVER_ERROR}],
+            )
+            return Response(
+                data=fallback_response.model_dump(mode="json"), 
+                status=status.HTTP_500_INTERNAL_SERVER_ERROR
+            )
+
+    def _handle_validation_errors(self, errors):
+        from todo.dto.responses.error_response import ApiErrorDetail, ApiErrorSource
+        
+        formatted_errors = []
+        for field, messages in errors.items():
+            if isinstance(messages, list):
+                for message in messages:
+                    formatted_errors.append(
+                        ApiErrorDetail(
+                            source={ApiErrorSource.PARAMETER: field},
+                            title=ApiErrors.VALIDATION_ERROR,
+                            detail=str(message),
+                        )
+                    )
+            else:
+                formatted_errors.append(
+                    ApiErrorDetail(
+                        source={ApiErrorSource.PARAMETER: field}, 
+                        title=ApiErrors.VALIDATION_ERROR, 
+                        detail=str(messages)
+                    )
+                )
+
+        error_response = ApiErrorResponse(
+            statusCode=400, 
+            message=ApiErrors.VALIDATION_ERROR, 
+            errors=formatted_errors
+        )
+
+        return Response(
+            data=error_response.model_dump(mode="json"), 
+            status=status.HTTP_400_BAD_REQUEST
+        )