Coverage for modules\erp\api.py: 40%

1from ninja import Router 

2from typing import List, Optional 

3from uuid import UUID 

4from core.auth import JWTAuth 

5from core.permissions import require_tenant 

6from core.exceptions import NotFoundException 

7from .schemas import ( 

8 AccountIn, AccountOut, 

9 ProductIn, ProductOut, 

10 StockMoveIn, StockMoveOut, 

11 InvoiceIn, InvoiceOut, InvoiceLineOut, InvoicePaymentIn, 

12 ShipmentIn, ShipmentOut, ShipmentUpdateIn 

13) 

14from .models import Account, Product, StockMove, Invoice, Warehouse, Shipment 

15from .services import ERPService 

16 

17router = Router(tags=["ERP & Accounting"]) 

18 

19def shipment_out(shipment): 

20 return ShipmentOut.model_validate(shipment) 

21 

22 

23@router.get("/shipments", auth=JWTAuth(), response={200: List[ShipmentOut]}) 

24def list_shipments(request, status: Optional[str] = None): 

25 tenant_id = require_tenant(request) 

26 qs = Shipment.objects.filter(tenant_id=tenant_id) 

27 if status: 

28 qs = qs.filter(status=status) 

29 return [shipment_out(shipment) for shipment in qs] 

30 

31 

32@router.post("/shipments", auth=JWTAuth(), response={201: ShipmentOut}) 

33def create_shipment(request, payload: ShipmentIn): 

34 tenant_id = require_tenant(request) 

35 shipment = ERPService.create_shipment(tenant_id, payload) 

36 return 201, shipment_out(shipment) 

37 

38 

39@router.get("/shipments/{shipment_id}", auth=JWTAuth(), response={200: ShipmentOut}) 

40def get_shipment(request, shipment_id: UUID): 

41 tenant_id = require_tenant(request) 

42 shipment = Shipment.objects.filter(id=shipment_id, tenant_id=tenant_id).first() 

43 if not shipment: 

44 raise NotFoundException("Shipment not found.") 

45 return shipment_out(shipment) 

46 

47 

48@router.patch("/shipments/{shipment_id}", auth=JWTAuth(), response={200: ShipmentOut}) 

49def update_shipment(request, shipment_id: UUID, payload: ShipmentUpdateIn): 

50 tenant_id = require_tenant(request) 

51 shipment = Shipment.objects.filter(id=shipment_id, tenant_id=tenant_id).first() 

52 if not shipment: 

53 raise NotFoundException("Shipment not found.") 

54 for field, value in payload.dict(exclude_unset=True).items(): 

55 setattr(shipment, field, value) 

56 shipment.save() 

57 return shipment_out(shipment) 

58 

59 

60@router.delete("/shipments/{shipment_id}", auth=JWTAuth(), response={204: None}) 

61def delete_shipment(request, shipment_id: UUID): 

62 tenant_id = require_tenant(request) 

63 deleted, _ = Shipment.objects.filter(id=shipment_id, tenant_id=tenant_id).delete() 

64 if not deleted: 

65 raise NotFoundException("Shipment not found.") 

66 return 204, None 

67 

68# ── Chart of Accounts ───────────────────────────────────────────────────────── 

69@router.get("/accounts", auth=JWTAuth(), response={200: List[AccountOut]}) 

70def list_accounts(request): 

71 tenant_id = require_tenant(request) 

72 ERPService.get_or_create_chart_of_accounts(tenant_id) 

73 accounts = Account.objects.filter(tenant_id=tenant_id) 

74 return list(accounts) 

75 

76@router.post("/accounts", auth=JWTAuth(), response={201: AccountOut}) 

77def create_account(request, payload: AccountIn): 

78 tenant_id = require_tenant(request) 

79 acct = Account.objects.create( 

80 tenant_id=tenant_id, 

81 **payload.dict() 

82 ) 

83 return 201, acct 

84 

85# ── Products & Inventory ────────────────────────────────────────────────────── 

86@router.get("/products", auth=JWTAuth(), response={200: List[ProductOut]}) 

87def list_products(request, search: Optional[str] = None): 

88 tenant_id = require_tenant(request) 

89 qs = Product.objects.filter(tenant_id=tenant_id) 

90 if search: 

91 qs = qs.filter(name__icontains=search) | qs.filter(sku__icontains=search) 

92 

93 result = [] 

94 for p in qs: 

95 result.append(ProductOut( 

96 id=p.id, 

97 sku=p.sku, 

98 name=p.name, 

99 description=p.description, 

100 category=p.category, 

101 cost_price=p.cost_price, 

102 selling_price=p.selling_price, 

103 tax_rate=p.tax_rate, 

104 unit_of_measure=p.unit_of_measure, 

105 is_active=p.is_active, 

106 current_stock=p.current_stock, 

107 reorder_level=p.reorder_level, 

108 created_at=p.created_at, 

109 updated_at=p.updated_at, 

110 )) 

111 return result 

112 

113@router.post("/products", auth=JWTAuth(), response={201: ProductOut}) 

114def create_product(request, payload: ProductIn): 

115 tenant_id = require_tenant(request) 

116 p = Product.objects.create( 

117 tenant_id=tenant_id, 

118 **payload.dict() 

119 ) 

120 return 201, ProductOut( 

121 id=p.id, 

122 sku=p.sku, 

123 name=p.name, 

124 description=p.description, 

125 category=p.category, 

126 cost_price=p.cost_price, 

127 selling_price=p.selling_price, 

128 tax_rate=p.tax_rate, 

129 unit_of_measure=p.unit_of_measure, 

130 is_active=p.is_active, 

131 current_stock=p.current_stock, 

132 reorder_level=p.reorder_level, 

133 created_at=p.created_at, 

134 updated_at=p.updated_at, 

135 ) 

136 

137# ── Stock Moves ─────────────────────────────────────────────────────────────── 

138@router.get("/stock-moves", auth=JWTAuth(), response={200: List[StockMoveOut]}) 

139def list_stock_moves(request, product_id: Optional[UUID] = None): 

140 tenant_id = require_tenant(request) 

141 qs = StockMove.objects.filter(tenant_id=tenant_id).select_related('product', 'source_warehouse', 'dest_warehouse') 

142 if product_id: 

143 qs = qs.filter(product_id=product_id) 

144 

145 result = [] 

146 for m in qs: 

147 result.append(StockMoveOut( 

148 id=m.id, 

149 product_id=m.product.id, 

150 product_name=m.product.name, 

151 product_sku=m.product.sku, 

152 source_warehouse_name=m.source_warehouse.name if m.source_warehouse else None, 

153 dest_warehouse_name=m.dest_warehouse.name if m.dest_warehouse else None, 

154 quantity=m.quantity, 

155 unit_cost=m.unit_cost, 

156 move_type=m.move_type, 

157 reference_doc=m.reference_doc, 

158 created_at=m.created_at, 

159 )) 

160 return result 

161 

162@router.post("/stock-moves", auth=JWTAuth(), response={201: StockMoveOut}) 

163def create_stock_move(request, payload: StockMoveIn): 

164 tenant_id = require_tenant(request) 

165 product = Product.objects.filter(id=payload.product_id, tenant_id=tenant_id).first() 

166 if not product: 

167 raise NotFoundException("Product not found.") 

168 

169 src_wh = Warehouse.objects.filter(id=payload.source_warehouse_id, tenant_id=tenant_id).first() if payload.source_warehouse_id else None 

170 dst_wh = Warehouse.objects.filter(id=payload.dest_warehouse_id, tenant_id=tenant_id).first() if payload.dest_warehouse_id else None 

171 

172 m = StockMove.objects.create( 

173 tenant_id=tenant_id, 

174 product=product, 

175 source_warehouse=src_wh, 

176 dest_warehouse=dst_wh, 

177 quantity=payload.quantity, 

178 unit_cost=payload.unit_cost, 

179 move_type=payload.move_type, 

180 reference_doc=payload.reference_doc, 

181 ) 

182 return 201, StockMoveOut( 

183 id=m.id, 

184 product_id=m.product.id, 

185 product_name=m.product.name, 

186 product_sku=m.product.sku, 

187 source_warehouse_name=m.source_warehouse.name if m.source_warehouse else None, 

188 dest_warehouse_name=m.dest_warehouse.name if m.dest_warehouse else None, 

189 quantity=m.quantity, 

190 unit_cost=m.unit_cost, 

191 move_type=m.move_type, 

192 reference_doc=m.reference_doc, 

193 created_at=m.created_at, 

194 ) 

195 

196# ── Invoices ────────────────────────────────────────────────────────────────── 

197@router.get("/invoices", auth=JWTAuth(), response={200: List[InvoiceOut]}) 

198def list_invoices(request, status: Optional[str] = None): 

199 tenant_id = require_tenant(request) 

200 qs = Invoice.objects.filter(tenant_id=tenant_id).prefetch_related('lines') 

201 if status: 

202 qs = qs.filter(status=status) 

203 

204 result = [] 

205 for inv in qs: 

206 result.append(InvoiceOut( 

207 id=inv.id, 

208 invoice_number=inv.invoice_number, 

209 customer_name=inv.customer_name, 

210 customer_email=inv.customer_email, 

211 deal_id=inv.deal_id, 

212 issue_date=inv.issue_date, 

213 due_date=inv.due_date, 

214 status=inv.status, 

215 currency=inv.currency, 

216 subtotal=inv.subtotal, 

217 tax_total=inv.tax_total, 

218 total_amount=inv.total_amount, 

219 amount_paid=inv.amount_paid, 

220 balance_due=inv.balance_due, 

221 notes=inv.notes, 

222 terms=inv.terms, 

223 lines=[ 

224 InvoiceLineOut( 

225 id=l.id, 

226 product_id=l.product_id, 

227 description=l.description, 

228 quantity=l.quantity, 

229 unit_price=l.unit_price, 

230 tax_rate=l.tax_rate, 

231 subtotal=l.subtotal, 

232 tax_amount=l.tax_amount, 

233 total=l.total, 

234 ) for l in inv.lines.all() 

235 ], 

236 created_at=inv.created_at, 

237 updated_at=inv.updated_at, 

238 )) 

239 return result 

240 

241@router.post("/invoices", auth=JWTAuth(), response={201: InvoiceOut}) 

242def create_invoice(request, payload: InvoiceIn): 

243 tenant_id = require_tenant(request) 

244 inv = ERPService.create_invoice(tenant_id, payload) 

245 return 201, InvoiceOut( 

246 id=inv.id, 

247 invoice_number=inv.invoice_number, 

248 customer_name=inv.customer_name, 

249 customer_email=inv.customer_email, 

250 deal_id=inv.deal_id, 

251 issue_date=inv.issue_date, 

252 due_date=inv.due_date, 

253 status=inv.status, 

254 currency=inv.currency, 

255 subtotal=inv.subtotal, 

256 tax_total=inv.tax_total, 

257 total_amount=inv.total_amount, 

258 amount_paid=inv.amount_paid, 

259 balance_due=inv.balance_due, 

260 notes=inv.notes, 

261 terms=inv.terms, 

262 lines=[ 

263 InvoiceLineOut( 

264 id=l.id, 

265 product_id=l.product_id, 

266 description=l.description, 

267 quantity=l.quantity, 

268 unit_price=l.unit_price, 

269 tax_rate=l.tax_rate, 

270 subtotal=l.subtotal, 

271 tax_amount=l.tax_amount, 

272 total=l.total, 

273 ) for l in inv.lines.all() 

274 ], 

275 created_at=inv.created_at, 

276 updated_at=inv.updated_at, 

277 ) 

278 

279@router.get("/invoices/{invoice_id}", auth=JWTAuth(), response={200: InvoiceOut}) 

280def get_invoice(request, invoice_id: UUID): 

281 tenant_id = require_tenant(request) 

282 inv = Invoice.objects.filter(id=invoice_id, tenant_id=tenant_id).prefetch_related('lines').first() 

283 if not inv: 

284 raise NotFoundException("Invoice not found.") 

285 return InvoiceOut( 

286 id=inv.id, 

287 invoice_number=inv.invoice_number, 

288 customer_name=inv.customer_name, 

289 customer_email=inv.customer_email, 

290 deal_id=inv.deal_id, 

291 issue_date=inv.issue_date, 

292 due_date=inv.due_date, 

293 status=inv.status, 

294 currency=inv.currency, 

295 subtotal=inv.subtotal, 

296 tax_total=inv.tax_total, 

297 total_amount=inv.total_amount, 

298 amount_paid=inv.amount_paid, 

299 balance_due=inv.balance_due, 

300 notes=inv.notes, 

301 terms=inv.terms, 

302 lines=[ 

303 InvoiceLineOut( 

304 id=l.id, 

305 product_id=l.product_id, 

306 description=l.description, 

307 quantity=l.quantity, 

308 unit_price=l.unit_price, 

309 tax_rate=l.tax_rate, 

310 subtotal=l.subtotal, 

311 tax_amount=l.tax_amount, 

312 total=l.total, 

313 ) for l in inv.lines.all() 

314 ], 

315 created_at=inv.created_at, 

316 updated_at=inv.updated_at, 

317 ) 

318 

319@router.post("/invoices/{invoice_id}/pay", auth=JWTAuth(), response={200: InvoiceOut}) 

320def record_payment(request, invoice_id: UUID, payload: InvoicePaymentIn): 

321 tenant_id = require_tenant(request) 

322 inv = ERPService.record_payment(tenant_id, invoice_id, payload.amount) 

323 return get_invoice(request, inv.id)