064d67f099
Add the Sales CRM service through collaboration, sync architecture/registry docs, expose crm.torbatyar.ir, and include a production deploy script. Co-authored-by: Cursor <cursoragent@cursor.com>
62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
"""CRM foundation repositories — Phase 6.0/6.1 + re-exports for sales engine."""
|
|
from __future__ import annotations
|
|
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.models.foundation import Contact, Lead, Organization, Quote, SalesActivity
|
|
from app.repositories.base import TenantBaseRepository
|
|
from app.repositories.sales_engine import ( # noqa: F401
|
|
OpportunityRepository,
|
|
PipelineRepository,
|
|
PipelineStageRepository,
|
|
)
|
|
|
|
|
|
class LeadRepository(TenantBaseRepository[Lead]):
|
|
model = Lead
|
|
|
|
async def get_by_lead_number(
|
|
self, tenant_id: UUID, lead_number: str
|
|
) -> Lead | None:
|
|
stmt = select(Lead).where(
|
|
Lead.tenant_id == tenant_id,
|
|
Lead.lead_number == lead_number,
|
|
Lead.is_deleted.is_(False),
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
class ContactRepository(TenantBaseRepository[Contact]):
|
|
model = Contact
|
|
|
|
|
|
class OrganizationRepository(TenantBaseRepository[Organization]):
|
|
model = Organization
|
|
|
|
async def get_by_name(self, tenant_id: UUID, name: str) -> Organization | None:
|
|
stmt = select(Organization).where(
|
|
Organization.tenant_id == tenant_id,
|
|
Organization.name == name,
|
|
Organization.is_deleted.is_(False),
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
class SalesActivityRepository(TenantBaseRepository[SalesActivity]):
|
|
model = SalesActivity
|
|
|
|
|
|
class QuoteRepository(TenantBaseRepository[Quote]):
|
|
model = Quote
|
|
|
|
async def get_by_number(self, tenant_id: UUID, quote_number: str) -> Quote | None:
|
|
stmt = select(Quote).where(
|
|
Quote.tenant_id == tenant_id, Quote.quote_number == quote_number
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|