leocrm-bot
4c7f09c518
- I001: import block sorting in main.py, routes/__init__.py, companies.py, conftest.py, check_indexes.py - F401: unused imports removed (asyncio, UTC, datetime, AsyncSession) - 6 ASYNC240/230 warnings remain (os.path.exists in async, open() in test) — minor, non-blocking
322 lines
11 KiB
Python
322 lines
11 KiB
Python
"""Company routes — full CRUD, FTS search, filter, pagination, export, N:M, emails."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.auth import check_permission
|
|
from app.core.db import get_db
|
|
from app.deps import get_current_user
|
|
from app.schemas.company import CompanyCreate, CompanyUpdate
|
|
from app.services import company_service
|
|
|
|
router = APIRouter(prefix="/api/v1/companies", tags=["companies"])
|
|
|
|
|
|
@router.get("")
|
|
async def list_companies(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
search: str | None = Query(None),
|
|
industry: str | None = Query(None),
|
|
sort_by: str = Query("name"),
|
|
sort_order: str = Query("asc", pattern="^(asc|desc)$"),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""List companies with pagination, FTS search, industry filter, and sorting."""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
result = await company_service.list_companies(
|
|
db,
|
|
tenant_id,
|
|
page=page,
|
|
page_size=page_size,
|
|
search=search,
|
|
industry=industry,
|
|
sort_by=sort_by,
|
|
sort_order=sort_order,
|
|
)
|
|
return result
|
|
|
|
|
|
@router.post("", status_code=status.HTTP_201_CREATED)
|
|
async def create_company(
|
|
body: CompanyCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Create a company. Requires write permission."""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
user_id = uuid.UUID(current_user["user_id"])
|
|
role = current_user.get("role", "viewer")
|
|
|
|
if not check_permission(role, "companies", "create"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail={"detail": "Insufficient permissions", "code": "forbidden"},
|
|
)
|
|
|
|
data = body.model_dump()
|
|
return await company_service.create_company(db, tenant_id, user_id, data)
|
|
|
|
|
|
@router.get("/export")
|
|
async def export_companies(
|
|
format: str = Query("csv", pattern="^(csv|xlsx)$"),
|
|
industry: str | None = Query(None),
|
|
search: str | None = Query(None),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Export companies as CSV (streaming) or XLSX (buffered)."""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
|
|
if format == "csv":
|
|
import csv
|
|
import io
|
|
|
|
from fastapi.responses import StreamingResponse
|
|
from sqlalchemy import select as sa_select
|
|
|
|
from app.core.db import get_session_factory
|
|
from app.models.company import Company
|
|
|
|
async def generate_csv():
|
|
"""Async generator yielding CSV rows (streaming, not buffered).
|
|
Creates its own DB session to avoid request-scoped session issues.
|
|
"""
|
|
output = io.StringIO()
|
|
writer = csv.writer(output)
|
|
header = [
|
|
"id", "name", "account_number", "industry", "phone",
|
|
"email", "website", "description", "created_at", "updated_at",
|
|
]
|
|
writer.writerow(header)
|
|
yield output.getvalue()
|
|
output.seek(0)
|
|
output.truncate(0)
|
|
|
|
batch_size = 500
|
|
offset = 0
|
|
factory = get_session_factory()
|
|
async with factory() as session:
|
|
while True:
|
|
q = sa_select(Company).where(
|
|
Company.tenant_id == tenant_id,
|
|
Company.deleted_at.is_(None),
|
|
)
|
|
if industry:
|
|
q = q.where(Company.industry == industry)
|
|
if search:
|
|
pattern = f"%{search}%"
|
|
q = q.where(Company.name.ilike(pattern))
|
|
q = q.order_by(Company.name.asc()).offset(offset).limit(batch_size)
|
|
result = await session.execute(q)
|
|
companies = result.scalars().all()
|
|
if not companies:
|
|
break
|
|
for c in companies:
|
|
writer.writerow([
|
|
str(c.id), c.name, c.account_number or "", c.industry or "",
|
|
c.phone or "", c.email or "", c.website or "",
|
|
c.description or "",
|
|
c.created_at.isoformat() if c.created_at else "",
|
|
c.updated_at.isoformat() if c.updated_at else "",
|
|
])
|
|
yield output.getvalue()
|
|
output.seek(0)
|
|
output.truncate(0)
|
|
offset += batch_size
|
|
|
|
return StreamingResponse(
|
|
generate_csv(),
|
|
media_type="text/csv",
|
|
headers={"Content-Disposition": "attachment; filename=companies.csv"},
|
|
)
|
|
elif format == "xlsx":
|
|
xlsx_data = await company_service.export_companies_xlsx(
|
|
db,
|
|
tenant_id,
|
|
industry=industry,
|
|
search=search,
|
|
)
|
|
return Response(
|
|
content=xlsx_data,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": "attachment; filename=companies.xlsx"},
|
|
)
|
|
raise HTTPException(400, detail={"detail": "Invalid format", "code": "invalid_format"})
|
|
|
|
|
|
@router.get("/{company_id}")
|
|
async def get_company(
|
|
company_id: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Get a single company with contacts array. Cross-tenant returns 404."""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
try:
|
|
cid = uuid.UUID(company_id)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
400, detail={"detail": "Invalid company_id", "code": "invalid_id"}
|
|
) from None
|
|
|
|
data = await company_service.get_company_detail(db, tenant_id, cid)
|
|
if data is None:
|
|
raise HTTPException(404, detail={"detail": "Company not found", "code": "not_found"})
|
|
return data
|
|
|
|
|
|
@router.put("/{company_id}")
|
|
async def update_company(
|
|
company_id: str,
|
|
body: CompanyUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Update a company (full PUT)."""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
user_id = uuid.UUID(current_user["user_id"])
|
|
role = current_user.get("role", "viewer")
|
|
|
|
if not check_permission(role, "companies", "update"):
|
|
raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
|
|
|
|
try:
|
|
cid = uuid.UUID(company_id)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
400, detail={"detail": "Invalid company_id", "code": "invalid_id"}
|
|
) from None
|
|
|
|
data = body.model_dump(exclude_unset=True)
|
|
result = await company_service.update_company(db, tenant_id, user_id, cid, data)
|
|
if result is None:
|
|
raise HTTPException(404, detail={"detail": "Company not found", "code": "not_found"})
|
|
return result
|
|
|
|
|
|
@router.delete("/{company_id}")
|
|
async def delete_company(
|
|
company_id: str,
|
|
cascade: bool = Query(False),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Soft-delete a company. Use cascade=true to also delete N:M links."""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
user_id = uuid.UUID(current_user["user_id"])
|
|
role = current_user.get("role", "viewer")
|
|
|
|
if not check_permission(role, "companies", "delete"):
|
|
raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
|
|
|
|
try:
|
|
cid = uuid.UUID(company_id)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
400, detail={"detail": "Invalid company_id", "code": "invalid_id"}
|
|
) from None
|
|
|
|
deleted = await company_service.soft_delete_company(
|
|
db, tenant_id, user_id, cid, cascade=cascade
|
|
)
|
|
if not deleted:
|
|
raise HTTPException(404, detail={"detail": "Company not found", "code": "not_found"})
|
|
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
|
|
|
|
|
@router.post("/{company_id}/contacts/{contact_id}")
|
|
async def link_company_contact(
|
|
company_id: str,
|
|
contact_id: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Link a contact to a company (N:M)."""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
user_id = uuid.UUID(current_user["user_id"])
|
|
role = current_user.get("role", "viewer")
|
|
|
|
if not check_permission(role, "companies", "update"):
|
|
raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
|
|
|
|
try:
|
|
comp_id = uuid.UUID(company_id)
|
|
cont_id = uuid.UUID(contact_id)
|
|
except ValueError:
|
|
raise HTTPException(400, detail={"detail": "Invalid ID", "code": "invalid_id"}) from None
|
|
|
|
result = await company_service.link_contact(db, tenant_id, user_id, comp_id, cont_id)
|
|
if result is None:
|
|
raise HTTPException(
|
|
404, detail={"detail": "Company or contact not found", "code": "not_found"}
|
|
)
|
|
return result
|
|
|
|
|
|
@router.delete("/{company_id}/contacts/{contact_id}")
|
|
async def unlink_company_contact(
|
|
company_id: str,
|
|
contact_id: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Unlink a contact from a company (N:M)."""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
user_id = uuid.UUID(current_user["user_id"])
|
|
role = current_user.get("role", "viewer")
|
|
|
|
if not check_permission(role, "companies", "update"):
|
|
raise HTTPException(403, detail={"detail": "Insufficient permissions", "code": "forbidden"})
|
|
|
|
try:
|
|
comp_id = uuid.UUID(company_id)
|
|
cont_id = uuid.UUID(contact_id)
|
|
except ValueError:
|
|
raise HTTPException(400, detail={"detail": "Invalid ID", "code": "invalid_id"}) from None
|
|
|
|
unlinked = await company_service.unlink_contact(db, tenant_id, user_id, comp_id, cont_id)
|
|
if not unlinked:
|
|
raise HTTPException(404, detail={"detail": "Link not found", "code": "not_found"})
|
|
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
|
|
|
|
|
@router.get("/{company_id}/emails")
|
|
async def get_company_emails(
|
|
company_id: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Get emails for a company (mail plugin inactive — returns empty array)."""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
try:
|
|
cid = uuid.UUID(company_id)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
400, detail={"detail": "Invalid company_id", "code": "invalid_id"}
|
|
) from None
|
|
|
|
# Verify company exists
|
|
from sqlalchemy import select
|
|
|
|
from app.models.company import Company
|
|
|
|
q = select(Company).where(
|
|
Company.id == cid,
|
|
Company.tenant_id == tenant_id,
|
|
Company.deleted_at.is_(None),
|
|
)
|
|
result = await db.execute(q)
|
|
if result.scalar_one_or_none() is None:
|
|
raise HTTPException(404, detail={"detail": "Company not found", "code": "not_found"})
|
|
|
|
return []
|