Migrate from using default case cost type to assigning cost type by cost model.

src/dispatch/case/enums.py

@@ -13,3 +13,10 @@ class CaseResolutionReason(DispatchEnum):
     user_acknowledge = "User Acknowledged"
     mitigated = "Mitigated"
     escalated = "Escalated"
+
+
+class CostModelType(DispatchEnum):
+    """Type of cost model used to calculate costs."""
+
+    new = "New"
+    classic = "Classic"

src/dispatch/case_cost/scheduled.py

@@ -6,15 +6,14 @@
 
 from dispatch.decorators import scheduled_project_task, timer
 from dispatch.case import service as case_service
-from dispatch.case.enums import CaseStatus
-from dispatch.case_cost_type import service as case_cost_type_service
+from dispatch.case.enums import CaseStatus, CostModelType
 from dispatch.project.models import Project
 from dispatch.scheduler import scheduler
 
 from .service import (
     calculate_case_response_cost,
     update_case_response_cost,
-    get_or_create_default_case_response_cost,
+    get_or_create_case_response_cost_by_model_type,
 )
 
 
@@ -26,25 +25,16 @@
 @scheduled_project_task
 def calculate_cases_response_cost(db_session: Session, project: Project):
     """Calculates and saves the response cost for all cases."""
-    response_cost_type = case_cost_type_service.get_default(
-        db_session=db_session, project_id=project.id
-    )
-
-    if response_cost_type is None:
-        log.warning(
-            f"A default cost type for response cost doesn't exist in the {project.name} project and organization {project.organization.name}. Response costs for cases won't be calculated."
-        )
-        return
-
     cases = case_service.get_all_by_status(
         db_session=db_session, project_id=project.id, statuses=[CaseStatus.new, CaseStatus.triage]
     )
 
     for case in cases:
         try:
             # we get the response cost for the given case
-            case_response_cost = get_or_create_default_case_response_cost(case, db_session)
-
+            case_response_cost = get_or_create_case_response_cost_by_model_type(
+                case=case, db_session=db_session, model_type=CostModelType.new
+            )
             # we don't need to update the cost of closed cases if they already have a response cost and this was updated after the case was closed
             if case.status == CaseStatus.closed:
                 if case_response_cost:

src/dispatch/case_cost/service.py

@@ -1,7 +1,7 @@
 from datetime import datetime, timedelta, timezone
 import logging
 import math
-from typing import List, Optional
+from typing import Optional
 
 from sqlalchemy.orm import Session
 
@@ -10,7 +10,7 @@
 from dispatch.case.models import Case
 from dispatch.case.type.models import CaseType
 from dispatch.case_cost_type import service as case_cost_type_service
-from dispatch.case_cost_type.models import CaseCostTypeRead
+from dispatch.case.enums import CostModelType
 from dispatch.participant import service as participant_service
 from dispatch.participant.models import ParticipantRead
 from dispatch.participant_activity import service as participant_activity_service
@@ -30,7 +30,7 @@ def get(*, db_session: Session, case_cost_id: int) -> Optional[CaseCost]:
     return db_session.query(CaseCost).filter(CaseCost.id == case_cost_id).one_or_none()
 
 
-def get_by_case_id(*, db_session: Session, case_id: int) -> List[Optional[CaseCost]]:
+def get_by_case_id(*, db_session, case_id: int) -> list[Optional[CaseCost]]:
     """Gets case costs by their case id."""
     return db_session.query(CaseCost).filter(CaseCost.case_id == case_id).all()
 
@@ -48,7 +48,7 @@ def get_by_case_id_and_case_cost_type_id(
     )
 
 
-def get_all(*, db_session: Session) -> List[Optional[CaseCost]]:
+def get_all(*, db_session: Session) -> list[Optional[CaseCost]]:
     """Gets all case costs."""
     return db_session.query(CaseCost)
 
@@ -117,60 +117,38 @@ def calculate_response_cost(hourly_rate, total_response_time_seconds) -> int:
     return math.ceil((total_response_time_seconds / SECONDS_IN_HOUR) * hourly_rate)
 
 
-def get_default_case_response_cost(case: Case, db_session: Session) -> Optional[CaseCost]:
-    response_cost_type = case_cost_type_service.get_default(
-        db_session=db_session, project_id=case.project.id
+def get_or_create_case_response_cost_by_model_type(
+    case: Case, model_type: str, db_session: Session
+) -> Optional[CaseCost]:
+    """Gets a case response cost for a specific model type."""
+    # Find the cost type matching the requested model type for the project
+    response_cost_type = case_cost_type_service.get_or_create_response_cost_type(
+        db_session=db_session, project_id=case.project.id, model_type=model_type
     )
 
     if not response_cost_type:
         log.warning(
-            f"A default cost type for response cost doesn't exist in the {case.project.name} project and organization {case.project.organization.name}. Response costs for case {case.name} won't be calculated."
+            f"A default cost type for model type {model_type} doesn't exist and could not be created in the {case.project.name} project. "
+            f"Response costs for case {case.name} won't be calculated for this model."
         )
         return None
 
-    return get_by_case_id_and_case_cost_type_id(
-        db_session=db_session,
-        case_id=case.id,
-        case_cost_type_id=response_cost_type.id,
+    # Retrieve or create the case cost for the given case and cost type
+    case_cost = get_by_case_id_and_case_cost_type_id(
+        db_session=db_session, case_id=case.id, case_cost_type_id=response_cost_type.id
     )
-
-
-def get_or_create_default_case_response_cost(case: Case, db_session: Session) -> Optional[CaseCost]:
-    """Gets or creates the default case cost for a case.
-
-    The default case cost is the cost associated with the participant effort in a case's response.
-    """
-    response_cost_type = case_cost_type_service.get_default(
-        db_session=db_session, project_id=case.project.id
-    )
-
-    if not response_cost_type:
-        log.warning(
-            f"A default cost type for response cost doesn't exist in the {case.project.name} project and organization {case.project.organization.name}. Response costs for case {case.name} won't be calculated."
+    if not case_cost:
+        case_cost = CaseCostCreate(
+            case=case, case_cost_type=response_cost_type, amount=0, project=case.project
         )
-        return None
+        case_cost = create(db_session=db_session, case_cost_in=case_cost)
 
-    case_response_cost = get_by_case_id_and_case_cost_type_id(
-        db_session=db_session,
-        case_id=case.id,
-        case_cost_type_id=response_cost_type.id,
-    )
-
-    if not case_response_cost:
-        # we create the response cost if it doesn't exist
-        case_cost_type = CaseCostTypeRead.from_orm(response_cost_type)
-        case_cost_in = CaseCostCreate(case_cost_type=case_cost_type, project=case.project)
-        case_response_cost = create(db_session=db_session, case_cost_in=case_cost_in)
-        case.case_costs.append(case_response_cost)
-        db_session.add(case)
-        db_session.commit()
-
-    return case_response_cost
+    return case_cost
 
 
 def fetch_case_events(
     case: Case, activity: CostModelActivity, oldest: str, db_session: Session
-) -> List[Optional[tuple[datetime.timestamp, str]]]:
+) -> list[Optional[tuple[datetime.timestamp, str]]]:
     """Fetches case events for a given case and cost model activity.
 
     Args:
@@ -180,7 +158,7 @@ def fetch_case_events(
         db_session: The database session.
 
     Returns:
-        List[Optional[tuple[datetime.timestamp, str]]]: A list of tuples containing the timestamp and user_id of each event.
+        list[Optional[tuple[datetime.timestamp, str]]]: A list of tuples containing the timestamp and user_id of each event.
     """
 
     plugin_instance = plugin_service.get_active_instance_by_slug(
@@ -239,7 +217,9 @@ def update_case_participant_activities(
     # Used for determining whether we've previously calculated the case cost.
     current_time = datetime.now(tz=timezone.utc).replace(tzinfo=None)
 
-    case_response_cost = get_or_create_default_case_response_cost(case=case, db_session=db_session)
+    case_response_cost = get_or_create_case_response_cost_by_model_type(
+        case=case, db_session=db_session, model_type=CostModelType.new
+    )
     if not case_response_cost:
         log.warning(
             f"Cannot calculate case response cost for case {case.name}. No default case response cost type created or found."
@@ -293,17 +273,16 @@ def calculate_case_response_cost(case: Case, db_session: Session) -> int:
     """Calculates the response cost of a given case."""
     # Iterate through all the listed activities and aggregate the total participant response time spent on the case.
     participants_total_response_time = timedelta(0)
-    particpant_activities = (
+    participant_activities = (
         participant_activity_service.get_all_case_participant_activities_for_case(
             db_session=db_session, case_id=case.id
         )
     )
-    for participant_activity in particpant_activities:
+    for participant_activity in participant_activities:
         participants_total_response_time += (
             participant_activity.ended_at - participant_activity.started_at
         )
 
-    # Calculate the cost based on the total participant response time.
     hourly_rate = get_hourly_rate(case.project)
     amount = calculate_response_cost(
         hourly_rate=hourly_rate,
@@ -315,29 +294,33 @@ def calculate_case_response_cost(case: Case, db_session: Session) -> int:
 def update_case_response_cost(case: Case, db_session: Session) -> int:
     """Updates the response cost of a given case.
 
-    This function logs all case participant activities since the last case cost update and recalculates the case response cost based on all logged participant activities.
+    This function logs all case participant activities since the last case cost update and
+    recalculates the case response cost using the new cost model.
 
     Args:
-        case_id: The case id.
+        case: The case to update costs for.
         db_session: The database session.
 
     Returns:
-        int: The case response cost in dollars.
+        dict[str, int]: Dictionary containing costs from both models {'new': new_cost, 'classic': classic_cost}
     """
-    # We update the case participant activities before calculating the case response cost
+    # Update case participant activities before calculating costs
     update_case_participant_activities(case=case, db_session=db_session)
-    amount = calculate_case_response_cost(case=case, db_session=db_session)
-    case_response_cost = get_or_create_default_case_response_cost(case=case, db_session=db_session)
 
-    if not case_response_cost:
-        log.warning(f"Cannot calculate case response cost for case {case.name}.")
-        return 0
+    results = {}
+
+    # Calculate and update new model cost
+    new_amount = calculate_case_response_cost(case=case, db_session=db_session)
+    new_cost = get_or_create_case_response_cost_by_model_type(
+        case=case, model_type=CostModelType.new, db_session=db_session
+    )
 
-    # We update the cost amount only if the case cost has changed
-    if case_response_cost.amount != amount:
-        case_response_cost.amount = amount
-        case.case_costs.append(case_response_cost)
-        db_session.add(case)
-        db_session.commit()
+    if new_cost:
+        if new_cost.amount != new_amount:
+            new_cost.amount = new_amount
+            case.case_costs.append(new_cost)
+            db_session.add(case)
+            db_session.commit()
+        results[CostModelType.new] = new_cost.amount
 
-    return case_response_cost.amount
+    return results

src/dispatch/case_cost_type/models.py

@@ -3,11 +3,10 @@
 from pydantic import Field
 
 from sqlalchemy import Column, Integer, String, Boolean
-from sqlalchemy.event import listen
 
 from sqlalchemy_utils import TSVectorType, JSONType
 
-from dispatch.database.core import Base, ensure_unique_default_per_project
+from dispatch.database.core import Base
 from dispatch.models import (
     DispatchBase,
     NameStr,
@@ -27,27 +26,24 @@ class CaseCostType(Base, TimeStampMixin, ProjectMixin):
     description = Column(String)
     category = Column(String)
     details = Column(JSONType, nullable=True)
-    default = Column(Boolean, default=False)
     editable = Column(Boolean, default=True)
+    model_type = Column(String, nullable=True)
 
     # full text search capabilities
     search_vector = Column(
         TSVectorType("name", "description", weights={"name": "A", "description": "B"})
     )
 
 
-listen(CaseCostType.default, "set", ensure_unique_default_per_project)
-
-
 # Pydantic Models
 class CaseCostTypeBase(DispatchBase):
     name: NameStr
     description: Optional[str] = Field(None, nullable=True)
     category: Optional[str] = Field(None, nullable=True)
     details: Optional[dict] = {}
     created_at: Optional[datetime]
-    default: Optional[bool]
     editable: Optional[bool]
+    model_type: Optional[str] = Field(None, nullable=False)
 
 
 class CaseCostTypeCreate(CaseCostTypeBase):

src/dispatch/case_cost_type/service.py

@@ -1,8 +1,9 @@
-from sqlalchemy.sql.expression import true
 from typing import List, Optional
 
+from dispatch.case.enums import CostModelType
 from dispatch.project import service as project_service
 
+from .config import default_case_cost_type
 from .models import (
     CaseCostType,
     CaseCostTypeCreate,
@@ -15,16 +16,65 @@ def get(*, db_session, case_cost_type_id: int) -> Optional[CaseCostType]:
     return db_session.query(CaseCostType).filter(CaseCostType.id == case_cost_type_id).one_or_none()
 
 
-def get_default(*, db_session, project_id: int) -> Optional[CaseCostType]:
-    """Returns the default case cost type."""
+def get_response_cost_type(
+    *, db_session, project_id: int, model_type: str
+) -> Optional[CaseCostType]:
+    """Gets the default response cost type."""
     return (
         db_session.query(CaseCostType)
-        .filter(CaseCostType.default == true())
         .filter(CaseCostType.project_id == project_id)
+        .filter(CaseCostType.model_type == model_type)
         .one_or_none()
     )
 
 
+def get_or_create_response_cost_type(
+    *, db_session, project_id: int, model_type: str = CostModelType.new
+) -> CaseCostType:
+    """Gets or creates the response case cost type."""
+    case_cost_type = get_response_cost_type(
+        db_session=db_session, project_id=project_id, model_type=model_type
+    )
+
+    if not case_cost_type:
+        case_cost_type_in = CaseCostTypeCreate(
+            name=default_case_cost_type["name"],
+            description=default_case_cost_type["description"],
+            category=default_case_cost_type["category"],
+            details=default_case_cost_type["details"],
+            editable=default_case_cost_type["editable"],
+            project=project_service.get(db_session=db_session, project_id=project_id),
+            model_type=model_type,
+        )
+        case_cost_type = create(db_session=db_session, case_cost_type_in=case_cost_type_in)
+
+    return case_cost_type
+
+
+def get_all_response_case_cost_types(
+    *, db_session, project_id: int
+) -> List[Optional[CaseCostType]]:
+    """Returns all response case cost types.
+
+    This function queries the database for all case cost types that are marked as the response cost type.
+    The following case cost types that match this description are:
+    - CaseCostType with model_type CLASSIC
+    - CaseCostType with model_type NEW
+
+    All other case types are not tied to the default response cost type.
+    """
+    return (
+        +db_session.query(CaseCostType)
+        .filter(CaseCostType.project_id == project_id)
+        .filter(CaseCostType.model_type == CostModelType.classic)
+        .one()
+        + db_session.query(CaseCostType)
+        .filter(CaseCostType.project_id == project_id)
+        .filter(CaseCostType.model_type == CostModelType.new)
+        .one()
+    )
+
+
 def get_by_name(*, db_session, project_id: int, case_cost_type_name: str) -> Optional[CaseCostType]:
     """Gets a case cost type by its name."""
     return (