File size: 4,956 Bytes
f8b03c1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | """
案件 API
API 层职责：
1. 接收 HTTP 请求，验证参数（通过 Schema）
2. 调用 Service 层方法
3. 返回响应
不包含：业务逻辑、权限检查、异常处理（依赖全局异常处理器）
"""
from __future__ import annotations
from typing import Any, cast
from django.http import HttpRequest
from ninja import Router
from apps.cases.schemas import CaseCreateFull, CaseFullOut, CaseIn, CaseOut, CaseUpdate
from apps.cases.services import CaseService
from apps.core.dto.request_context import extract_request_context
router = Router()
def _get_case_service() -> CaseService:
from apps.contracts.services.contract.wiring import get_contract_service
return CaseService(contract_service=get_contract_service())
def _get_case_query_facade() -> CaseService:
return _get_case_service()
def _get_case_mutation_facade() -> CaseService:
return _get_case_service()
@router.get("/cases/search", response=list[CaseOut])
def search_cases(
request: HttpRequest,
q: str,
limit: int | None = 10,
) -> list[CaseOut]:
"""搜索案件"""
service = _get_case_query_facade()
ctx = extract_request_context(request)
return cast(
list[CaseOut],
service.search_cases(
query=q,
limit=limit, # type: ignore[arg-type]
user=ctx.user,
org_access=ctx.org_access,
perm_open_access=ctx.perm_open_access,
),
)
@router.get("/cases", response=list[CaseOut])
def list_cases(
request: HttpRequest,
case_type: str | None = None,
status: str | None = None,
case_number: str | None = None,
) -> list[CaseOut]:
"""获取案件列表"""
service = _get_case_query_facade()
ctx = extract_request_context(request)
if case_number:
return cast(
list[CaseOut],
service.search_by_case_number(
case_number=case_number,
user=ctx.user,
org_access=ctx.org_access,
perm_open_access=ctx.perm_open_access,
),
)
return cast(
list[CaseOut],
service.list_cases(
case_type=case_type,
status=status,
user=ctx.user,
org_access=ctx.org_access,
perm_open_access=ctx.perm_open_access,
),
)
@router.get("/cases/{case_id}", response=CaseOut)
def get_case(request: HttpRequest, case_id: int) -> CaseOut:
"""获取单个案件"""
service = _get_case_query_facade()
ctx = extract_request_context(request)
return cast(
CaseOut,
service.get_case(
case_id=case_id,
user=ctx.user,
org_access=ctx.org_access,
perm_open_access=ctx.perm_open_access,
),
)
@router.post("/cases", response=CaseOut)
def create_case(request: HttpRequest, payload: CaseIn) -> CaseOut:
"""创建案件"""
service = _get_case_mutation_facade()
ctx = extract_request_context(request)
data = payload.dict()
return cast(CaseOut, service.create_case(data, user=ctx.user))
@router.put("/cases/{case_id}", response=CaseOut)
def update_case(request: HttpRequest, case_id: int, payload: CaseUpdate) -> CaseOut:
"""更新案件"""
service = _get_case_mutation_facade()
ctx = extract_request_context(request)
data = payload.dict(exclude_unset=True)
return cast(CaseOut, service.update_case(case_id, data, user=ctx.user))
@router.delete("/cases/{case_id}")
def delete_case(request: HttpRequest, case_id: int) -> dict[str, bool]:
"""删除案件"""
service = _get_case_mutation_facade()
ctx = extract_request_context(request)
service.delete_case(case_id, user=ctx.user)
return {"success": True}
@router.post("/cases/full", response=CaseFullOut)
def create_case_full(request: HttpRequest, payload: CaseCreateFull) -> CaseFullOut:
"""创建完整案件（包含当事人、指派、日志）"""
service = _get_case_mutation_facade()
ctx = extract_request_context(request)
actor_id = getattr(ctx.user, "id", None) if ctx.user else None
data: dict[str, Any] = {
"case": payload.case.dict(),
"parties": [p.dict() for p in payload.parties] if payload.parties else [],
"assignments": [a.dict() for a in payload.assignments] if payload.assignments else [],
"logs": [log.dict() for log in payload.logs] if payload.logs else [],
"supervising_authorities": (
[s.dict() for s in payload.supervising_authorities] if payload.supervising_authorities else []
),
}
result = service.create_case_full(data, actor_id=actor_id, user=ctx.user)
return CaseFullOut(
case=result["case"],
parties=result["parties"],
assignments=result["assignments"],
logs=result["logs"],
case_numbers=[],
supervising_authorities=result.get("supervising_authorities", []),
)
|