def create()

in src/dispatch/case/service.py [0:0]

• 77 lines of code
• 14 McCabe index (conditional complexity)

def create(*, db_session, case_in: CaseCreate, current_user: DispatchUser = None) -> Case:
    """Creates a new case.

    Returns:
        The created case.

    Raises:
        ValidationError: If the case type does not have a conversation target and the case is not being created with a dedicated channel, the case will not be created.
    """
    project = project_service.get_by_name_or_default(
        db_session=db_session, project_in=case_in.project
    )

    tag_objs = []
    for t in case_in.tags:
        tag_objs.append(tag_service.get_or_create(db_session=db_session, tag_in=t))

    # TODO(mvilanova): allow to provide related cases and incidents, and duplicated cases
    case_type = case_type_service.get_by_name_or_default(
        db_session=db_session, project_id=project.id, case_type_in=case_in.case_type
    )

    # Cases with dedicated channels do not require a conversation target.
    if not case_in.dedicated_channel:
        if not case_type or not case_type.conversation_target:
            raise ValueError(
                f"Cases without dedicated channels require a conversation target. Case type with name {case_in.case_type.name} does not have a conversation target. The case will not be created."
            )

    case = Case(
        title=case_in.title,
        description=case_in.description,
        project=project,
        status=case_in.status,
        dedicated_channel=case_in.dedicated_channel,
        tags=tag_objs,
        case_type=case_type,
        event=case_in.event,
    )

    case.visibility = case_type.visibility
    if case_in.visibility:
        case.visibility = case_in.visibility

    case_severity = case_severity_service.get_by_name_or_default(
        db_session=db_session, project_id=project.id, case_severity_in=case_in.case_severity
    )
    case.case_severity = case_severity

    case_priority = case_priority_service.get_by_name_or_default(
        db_session=db_session, project_id=project.id, case_priority_in=case_in.case_priority
    )
    case.case_priority = case_priority

    db_session.add(case)
    db_session.commit()

    event_service.log_case_event(
        db_session=db_session,
        source="Dispatch Core App",
        description="Case created",
        details={
            "title": case.title,
            "description": case.description,
            "type": case.case_type.name,
            "severity": case.case_severity.name,
            "priority": case.case_priority.name,
            "status": case.status,
            "visibility": case.visibility,
        },
        case_id=case.id,
    )

    assignee_email = None
    if case_in.assignee:
        # we assign the case to the assignee provided
        assignee_email = case_in.assignee.individual.email
    elif case_type.oncall_service:
        # we assign the case to the oncall person for the given case type
        assignee_email = service_flows.resolve_oncall(
            service=case_type.oncall_service, db_session=db_session
        )
    elif current_user:
        # we fall back to assign the case to the current user
        assignee_email = current_user.email

    # add assignee
    if assignee_email:
        participant_flows.add_participant(
            assignee_email,
            case,
            db_session,
            roles=[ParticipantRoleType.assignee],
        )

    # add reporter
    if case_in.reporter:
        participant_flows.add_participant(
            case_in.reporter.individual.email,
            case,
            db_session,
            roles=[ParticipantRoleType.reporter],
        )

    return case