ss-tools/backend/src/api/routes/git/_merge_routes.py

# #region GitMergeRoutes [C:3] [TYPE Module] [SEMANTICS git, merge, routes, conflicts]
# @BRIEF FastAPI endpoints for merge operations (status, conflicts, resolve, abort, continue).
# @LAYER: API

from typing import List, Optional

from fastapi import Depends, HTTPException

from src.core.logger import belief_scope
from src.dependencies import get_config_manager, has_permission

from src.api.routes.git_schemas import (
    MergeConflictFileSchema,
    MergeContinueRequest,
    MergeResolveRequest,
    MergeStatusSchema,
)

from ._router import router
from ._helpers import (
    _handle_unexpected_git_route_error,
)
from ._deps import get_git_service


# #region get_merge_status [C:2] [TYPE Function]
# @BRIEF Return unfinished-merge status for repository (web-only recovery support).
@router.get(
    "/repositories/{dashboard_ref}/merge/status", response_model=MergeStatusSchema
)
async def get_merge_status(
    dashboard_ref: str,
    env_id: Optional[str] = None,
    config_manager=Depends(get_config_manager),
    _=Depends(has_permission("plugin:git", "EXECUTE")),
):
    _gs = get_git_service()
    with belief_scope("get_merge_status"):
        from . import _resolve_dashboard_id_from_ref

        try:
            dashboard_id = _resolve_dashboard_id_from_ref(
                dashboard_ref, config_manager, env_id
            )
            return _gs.get_merge_status(dashboard_id)
        except HTTPException:
            raise
        except Exception as e:
            _handle_unexpected_git_route_error("get_merge_status", e)
# #endregion get_merge_status


# #region get_merge_conflicts [C:2] [TYPE Function]
# @BRIEF Return conflicted files with mine/theirs previews for web conflict resolver.
@router.get(
    "/repositories/{dashboard_ref}/merge/conflicts",
    response_model=List[MergeConflictFileSchema],
)
async def get_merge_conflicts(
    dashboard_ref: str,
    env_id: Optional[str] = None,
    config_manager=Depends(get_config_manager),
    _=Depends(has_permission("plugin:git", "EXECUTE")),
):
    _gs = get_git_service()
    with belief_scope("get_merge_conflicts"):
        from . import _resolve_dashboard_id_from_ref

        try:
            dashboard_id = _resolve_dashboard_id_from_ref(
                dashboard_ref, config_manager, env_id
            )
            return _gs.get_merge_conflicts(dashboard_id)
        except HTTPException:
            raise
        except Exception as e:
            _handle_unexpected_git_route_error("get_merge_conflicts", e)
# #endregion get_merge_conflicts


# #region resolve_merge_conflicts [C:3] [TYPE Function]
# @BRIEF Apply mine/theirs/manual conflict resolutions from WebUI and stage files.
# @RELATION CALLS -> [GitService]
@router.post("/repositories/{dashboard_ref}/merge/resolve")
async def resolve_merge_conflicts(
    dashboard_ref: str,
    resolve_data: MergeResolveRequest,
    env_id: Optional[str] = None,
    config_manager=Depends(get_config_manager),
    _=Depends(has_permission("plugin:git", "EXECUTE")),
):
    _gs = get_git_service()
    with belief_scope("resolve_merge_conflicts"):
        from . import _resolve_dashboard_id_from_ref

        try:
            dashboard_id = _resolve_dashboard_id_from_ref(
                dashboard_ref, config_manager, env_id
            )
            resolved_files = _gs.resolve_merge_conflicts(
                dashboard_id,
                [item.dict() for item in resolve_data.resolutions],
            )
            return {"status": "success", "resolved_files": resolved_files}
        except HTTPException:
            raise
        except Exception as e:
            _handle_unexpected_git_route_error("resolve_merge_conflicts", e)
# #endregion resolve_merge_conflicts


# #region abort_merge [C:2] [TYPE Function]
# @BRIEF Abort unfinished merge from WebUI flow.
@router.post("/repositories/{dashboard_ref}/merge/abort")
async def abort_merge(
    dashboard_ref: str,
    env_id: Optional[str] = None,
    config_manager=Depends(get_config_manager),
    _=Depends(has_permission("plugin:git", "EXECUTE")),
):
    _gs = get_git_service()
    with belief_scope("abort_merge"):
        from . import _resolve_dashboard_id_from_ref

        try:
            dashboard_id = _resolve_dashboard_id_from_ref(
                dashboard_ref, config_manager, env_id
            )
            return _gs.abort_merge(dashboard_id)
        except HTTPException:
            raise
        except Exception as e:
            _handle_unexpected_git_route_error("abort_merge", e)
# #endregion abort_merge


# #region continue_merge [C:3] [TYPE Function]
# @BRIEF Finalize unfinished merge from WebUI flow.
# @RELATION CALLS -> [GitService]
@router.post("/repositories/{dashboard_ref}/merge/continue")
async def continue_merge(
    dashboard_ref: str,
    continue_data: MergeContinueRequest,
    env_id: Optional[str] = None,
    config_manager=Depends(get_config_manager),
    _=Depends(has_permission("plugin:git", "EXECUTE")),
):
    _gs = get_git_service()
    with belief_scope("continue_merge"):
        from . import _resolve_dashboard_id_from_ref

        try:
            dashboard_id = _resolve_dashboard_id_from_ref(
                dashboard_ref, config_manager, env_id
            )
            return _gs.continue_merge(dashboard_id, continue_data.message)
        except HTTPException:
            raise
        except Exception as e:
            _handle_unexpected_git_route_error("continue_merge", e)
# #endregion continue_merge
# #endregion GitMergeRoutes