Coverage for modules\customs\api.py: 42%

1from ninja import Router 

2from typing import List, Optional 

3from uuid import UUID 

4from core.auth import JWTAuth 

5from core.permissions import require_tenant, require_admin 

6from core.exceptions import NotFoundException 

7from .schemas import ( 

8 CustomsEntryIn, CustomsEntryOut, 

9 ComplianceDocumentIn, ComplianceDocumentOut, 

10 DutyRateIn, DutyRateOut, 

11 BrokerLicenseIn, BrokerLicenseOut, 

12) 

13from .models import CustomsEntry, ComplianceDocument, DutyRate, BrokerLicense 

14 

15router = Router(tags=["Customs"]) 

16 

17# ── Customs Entries ──────────────────────────────────────────────────────────── 

18@router.get("/entries", auth=JWTAuth(), response={200: List[CustomsEntryOut]}) 

19def list_customs_entries(request, status: Optional[str] = None, entry_type: Optional[str] = None): 

20 tenant_id = require_tenant(request) 

21 qs = CustomsEntry.objects.filter(tenant_id=tenant_id) 

22 if status: 

23 qs = qs.filter(status=status) 

24 if entry_type: 

25 qs = qs.filter(entry_type=entry_type) 

26 return list(qs) 

27 

28 

29@router.post("/entries", auth=JWTAuth(), response={201: CustomsEntryOut}) 

30@require_admin 

31def create_customs_entry(request, payload: CustomsEntryIn): 

32 tenant_id = require_tenant(request) 

33 entry = CustomsEntry.objects.create(tenant_id=tenant_id, **payload.dict()) 

34 return 201, entry 

35 

36 

37@router.get("/entries/{entry_id}", auth=JWTAuth(), response={200: CustomsEntryOut}) 

38def get_customs_entry(request, entry_id: UUID): 

39 tenant_id = require_tenant(request) 

40 entry = CustomsEntry.objects.filter(id=entry_id, tenant_id=tenant_id).first() 

41 if not entry: 

42 raise NotFoundException("Customs entry not found.") 

43 return entry 

44 

45 

46@router.put("/entries/{entry_id}", auth=JWTAuth(), response={200: CustomsEntryOut}) 

47@require_admin 

48def update_customs_entry(request, entry_id: UUID, payload: CustomsEntryIn): 

49 tenant_id = require_tenant(request) 

50 entry = CustomsEntry.objects.filter(id=entry_id, tenant_id=tenant_id).first() 

51 if not entry: 

52 raise NotFoundException("Customs entry not found.") 

53 for key, value in payload.dict().items(): 

54 setattr(entry, key, value) 

55 entry.save() 

56 return entry 

57 

58 

59@router.delete("/entries/{entry_id}", auth=JWTAuth()) 

60@require_admin 

61def delete_customs_entry(request, entry_id: UUID): 

62 tenant_id = require_tenant(request) 

63 entry = CustomsEntry.objects.filter(id=entry_id, tenant_id=tenant_id).first() 

64 if not entry: 

65 raise NotFoundException("Customs entry not found.") 

66 entry.delete() 

67 return {"success": True} 

68 

69 

70# ── Compliance Documents ─────────────────────────────────────────────────────── 

71@router.get("/documents", auth=JWTAuth(), response={200: List[ComplianceDocumentOut]}) 

72def list_documents(request, entry_id: Optional[UUID] = None): 

73 tenant_id = require_tenant(request) 

74 qs = ComplianceDocument.objects.filter(tenant_id=tenant_id) 

75 if entry_id: 

76 qs = qs.filter(entry_id=entry_id) 

77 return list(qs.select_related("entry")) 

78 

79 

80@router.post("/documents", auth=JWTAuth(), response={201: ComplianceDocumentOut}) 

81@require_admin 

82def create_document(request, payload: ComplianceDocumentIn): 

83 tenant_id = require_tenant(request) 

84 doc = ComplianceDocument.objects.create(tenant_id=tenant_id, **payload.dict()) 

85 return 201, doc 

86 

87 

88@router.delete("/documents/{doc_id}", auth=JWTAuth()) 

89@require_admin 

90def delete_document(request, doc_id: UUID): 

91 tenant_id = require_tenant(request) 

92 doc = ComplianceDocument.objects.filter(id=doc_id, tenant_id=tenant_id).first() 

93 if not doc: 

94 raise NotFoundException("Document not found.") 

95 doc.delete() 

96 return {"success": True} 

97 

98 

99# ── Duty Rates ──────────────────────────────────────────────────────────────── 

100@router.get("/duty-rates", auth=JWTAuth(), response={200: List[DutyRateOut]}) 

101def list_duty_rates(request, hs_code: Optional[str] = None): 

102 tenant_id = require_tenant(request) 

103 qs = DutyRate.objects.filter(tenant_id=tenant_id, is_active=True) 

104 if hs_code: 

105 qs = qs.filter(hs_code__startswith=hs_code) 

106 return list(qs) 

107 

108 

109@router.post("/duty-rates", auth=JWTAuth(), response={201: DutyRateOut}) 

110@require_admin 

111def create_duty_rate(request, payload: DutyRateIn): 

112 tenant_id = require_tenant(request) 

113 rate = DutyRate.objects.create(tenant_id=tenant_id, **payload.dict()) 

114 return 201, rate 

115 

116 

117# ── Broker Licenses ─────────────────────────────────────────────────────────── 

118@router.get("/broker-licenses", auth=JWTAuth(), response={200: List[BrokerLicenseOut]}) 

119def list_broker_licenses(request): 

120 tenant_id = require_tenant(request) 

121 return list(BrokerLicense.objects.filter(tenant_id=tenant_id, is_active=True)) 

122 

123 

124@router.post("/broker-licenses", auth=JWTAuth(), response={201: BrokerLicenseOut}) 

125@require_admin 

126def create_broker_license(request, payload: BrokerLicenseIn): 

127 tenant_id = require_tenant(request) 

128 license = BrokerLicense.objects.create(tenant_id=tenant_id, **payload.dict()) 

129 return 201, license