Coverage for modules\crm\api.py: 34%
93 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-05 09:25 +0800
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 CompanyIn, CompanyOut,
9 ContactIn, ContactOut,
10 PipelineOut, DealStageOut,
11 DealIn, DealUpdateIn, DealOut,
12 ActivityIn, ActivityOut
13)
14from .models import Company, Contact, Pipeline, DealStage, Deal, Activity
15from .services import CRMService
17router = Router(tags=["CRM"])
19# ── Companies ─────────────────────────────────────────────────────────────────
20@router.get("/companies", auth=JWTAuth(), response={200: List[CompanyOut]})
21def list_companies(request, search: Optional[str] = None):
22 tenant_id = require_tenant(request)
23 qs = Company.objects.filter(tenant_id=tenant_id)
24 if search:
25 qs = qs.filter(name__icontains=search)
26 return list(qs)
28@router.post("/companies", auth=JWTAuth(), response={201: CompanyOut})
29def create_company(request, payload: CompanyIn):
30 tenant_id = require_tenant(request)
31 company = Company.objects.create(
32 tenant_id=tenant_id,
33 **payload.dict()
34 )
35 return 201, company
37@router.get("/companies/{company_id}", auth=JWTAuth(), response={200: CompanyOut})
38def get_company(request, company_id: UUID):
39 tenant_id = require_tenant(request)
40 company = Company.objects.filter(id=company_id, tenant_id=tenant_id).first()
41 if not company:
42 raise NotFoundException("Company not found.")
43 return company
45# ── Contacts ──────────────────────────────────────────────────────────────────
46@router.get("/contacts", auth=JWTAuth(), response={200: List[ContactOut]})
47def list_contacts(request, company_id: Optional[UUID] = None, search: Optional[str] = None):
48 tenant_id = require_tenant(request)
49 qs = Contact.objects.filter(tenant_id=tenant_id).select_related('company')
50 if company_id:
51 qs = qs.filter(company_id=company_id)
52 if search:
53 qs = qs.filter(first_name__icontains=search) | qs.filter(last_name__icontains=search) | qs.filter(email__icontains=search)
55 result = []
56 for c in qs:
57 result.append(ContactOut(
58 id=c.id,
59 first_name=c.first_name,
60 last_name=c.last_name,
61 full_name=c.full_name,
62 email=c.email,
63 phone=c.phone,
64 job_title=c.job_title,
65 company_id=c.company.id if c.company else None,
66 company_name=c.company.name if c.company else None,
67 is_primary=c.is_primary,
68 avatar_url=c.avatar_url,
69 created_at=c.created_at,
70 updated_at=c.updated_at,
71 ))
72 return result
74@router.post("/contacts", auth=JWTAuth(), response={201: ContactOut})
75def create_contact(request, payload: ContactIn):
76 tenant_id = require_tenant(request)
77 company = Company.objects.filter(id=payload.company_id, tenant_id=tenant_id).first() if payload.company_id else None
78 c = Contact.objects.create(
79 tenant_id=tenant_id,
80 first_name=payload.first_name,
81 last_name=payload.last_name or "",
82 email=payload.email,
83 phone=payload.phone,
84 job_title=payload.job_title,
85 company=company,
86 is_primary=payload.is_primary,
87 avatar_url=payload.avatar_url,
88 )
89 return 201, ContactOut(
90 id=c.id,
91 first_name=c.first_name,
92 last_name=c.last_name,
93 full_name=c.full_name,
94 email=c.email,
95 phone=c.phone,
96 job_title=c.job_title,
97 company_id=c.company.id if c.company else None,
98 company_name=c.company.name if c.company else None,
99 is_primary=c.is_primary,
100 avatar_url=c.avatar_url,
101 created_at=c.created_at,
102 updated_at=c.updated_at,
103 )
105# ── Pipelines & Deals ─────────────────────────────────────────────────────────
106@router.get("/pipelines", auth=JWTAuth(), response={200: List[PipelineOut]})
107def list_pipelines(request):
108 tenant_id = require_tenant(request)
109 # Ensure default pipeline exists
110 CRMService.get_or_create_default_pipeline(tenant_id)
111 pipelines = Pipeline.objects.filter(tenant_id=tenant_id).prefetch_related('stages')
112 result = []
113 for p in pipelines:
114 result.append(PipelineOut(
115 id=p.id,
116 name=p.name,
117 is_default=p.is_default,
118 stages=[
119 DealStageOut(
120 id=s.id,
121 name=s.name,
122 order=s.order,
123 probability=s.probability,
124 color=s.color,
125 ) for s in p.stages.all()
126 ]
127 ))
128 return result
130@router.get("/deals", auth=JWTAuth(), response={200: List[DealOut]})
131def list_deals(request, pipeline_id: Optional[UUID] = None, stage_id: Optional[UUID] = None):
132 tenant_id = require_tenant(request)
133 qs = Deal.objects.filter(tenant_id=tenant_id).select_related('stage', 'contact', 'company', 'assigned_to')
134 if pipeline_id:
135 qs = qs.filter(pipeline_id=pipeline_id)
136 if stage_id:
137 qs = qs.filter(stage_id=stage_id)
139 result = []
140 for d in qs:
141 result.append(DealOut(
142 id=d.id,
143 title=d.title,
144 value=d.value,
145 currency=d.currency,
146 status=d.status,
147 pipeline_id=d.pipeline_id,
148 stage_id=d.stage.id,
149 stage_name=d.stage.name,
150 contact_id=d.contact.id if d.contact else None,
151 contact_name=d.contact.full_name if d.contact else None,
152 company_id=d.company.id if d.company else None,
153 company_name=d.company.name if d.company else None,
154 expected_close_date=d.expected_close_date,
155 closed_at=d.closed_at,
156 assigned_to_id=d.assigned_to.id if d.assigned_to else None,
157 assigned_to_name=d.assigned_to.full_name if d.assigned_to else None,
158 created_at=d.created_at,
159 updated_at=d.updated_at,
160 ))
161 return result
163@router.post("/deals", auth=JWTAuth(), response={201: DealOut})
164def create_deal(request, payload: DealIn):
165 tenant_id = require_tenant(request)
166 d = CRMService.create_deal(tenant_id, payload)
167 return 201, DealOut(
168 id=d.id,
169 title=d.title,
170 value=d.value,
171 currency=d.currency,
172 status=d.status,
173 pipeline_id=d.pipeline_id,
174 stage_id=d.stage.id,
175 stage_name=d.stage.name,
176 contact_id=d.contact.id if d.contact else None,
177 contact_name=d.contact.full_name if d.contact else None,
178 company_id=d.company.id if d.company else None,
179 company_name=d.company.name if d.company else None,
180 expected_close_date=d.expected_close_date,
181 closed_at=d.closed_at,
182 assigned_to_id=d.assigned_to.id if d.assigned_to else None,
183 assigned_to_name=d.assigned_to.full_name if d.assigned_to else None,
184 created_at=d.created_at,
185 updated_at=d.updated_at,
186 )
188@router.patch("/deals/{deal_id}", auth=JWTAuth(), response={200: DealOut})
189def update_deal(request, deal_id: UUID, payload: DealUpdateIn):
190 tenant_id = require_tenant(request)
191 d = CRMService.update_deal(tenant_id, deal_id, payload)
192 return DealOut(
193 id=d.id,
194 title=d.title,
195 value=d.value,
196 currency=d.currency,
197 status=d.status,
198 pipeline_id=d.pipeline_id,
199 stage_id=d.stage.id,
200 stage_name=d.stage.name,
201 contact_id=d.contact.id if d.contact else None,
202 contact_name=d.contact.full_name if d.contact else None,
203 company_id=d.company.id if d.company else None,
204 company_name=d.company.name if d.company else None,
205 expected_close_date=d.expected_close_date,
206 closed_at=d.closed_at,
207 assigned_to_id=d.assigned_to.id if d.assigned_to else None,
208 assigned_to_name=d.assigned_to.full_name if d.assigned_to else None,
209 created_at=d.created_at,
210 updated_at=d.updated_at,
211 )
213# ── Activities ────────────────────────────────────────────────────────────────
214@router.get("/activities", auth=JWTAuth(), response={200: List[ActivityOut]})
215def list_activities(request, target_type: Optional[str] = None, target_id: Optional[UUID] = None):
216 tenant_id = require_tenant(request)
217 qs = Activity.objects.filter(tenant_id=tenant_id).select_related('created_by', 'assigned_to')
218 if target_type and target_id:
219 qs = qs.filter(target_type=target_type, target_id=target_id)
221 result = []
222 for a in qs:
223 result.append(ActivityOut(
224 id=a.id,
225 target_type=a.target_type,
226 target_id=a.target_id,
227 activity_type=a.activity_type,
228 subject=a.subject,
229 body=a.body,
230 due_date=a.due_date,
231 completed_at=a.completed_at,
232 assigned_to_id=a.assigned_to.id if a.assigned_to else None,
233 assigned_to_name=a.assigned_to.full_name if a.assigned_to else None,
234 created_by_id=a.created_by.id,
235 created_by_name=a.created_by.full_name,
236 created_at=a.created_at,
237 ))
238 return result
240@router.post("/activities", auth=JWTAuth(), response={201: ActivityOut})
241def create_activity(request, payload: ActivityIn):
242 tenant_id = require_tenant(request)
243 a = Activity.objects.create(
244 tenant_id=tenant_id,
245 target_type=payload.target_type,
246 target_id=payload.target_id,
247 activity_type=payload.activity_type,
248 subject=payload.subject,
249 body=payload.body,
250 due_date=payload.due_date,
251 assigned_to_id=payload.assigned_to_id,
252 created_by=request.user,
253 )
254 return 201, ActivityOut(
255 id=a.id,
256 target_type=a.target_type,
257 target_id=a.target_id,
258 activity_type=a.activity_type,
259 subject=a.subject,
260 body=a.body,
261 due_date=a.due_date,
262 completed_at=a.completed_at,
263 assigned_to_id=a.assigned_to.id if a.assigned_to else None,
264 assigned_to_name=a.assigned_to.full_name if a.assigned_to else None,
265 created_by_id=a.created_by.id,
266 created_by_name=a.created_by.full_name,
267 created_at=a.created_at,
268 )