mirror of
https://github.com/idrainformatica/PecFlow.git
synced 2026-06-16 12:45:42 +02:00
133 lines
4.4 KiB
Python
133 lines
4.4 KiB
Python
"""
|
||
Router rubrica indirizzi PEC (Feature 6).
|
||
|
||
Endpoint:
|
||
GET /contacts – lista contatti (con ricerca)
|
||
POST /contacts – crea contatto manuale
|
||
GET /contacts/autocomplete – autocomplete per compose
|
||
GET /contacts/{id} – dettaglio contatto
|
||
PUT /contacts/{id} – aggiorna contatto
|
||
DELETE /contacts/{id} – elimina contatto
|
||
POST /contacts/import – importa da CSV
|
||
"""
|
||
|
||
import uuid
|
||
|
||
from fastapi import APIRouter, Query, UploadFile, File, status
|
||
|
||
from app.dependencies import CurrentUser, DB
|
||
from app.schemas.pec_contact import (
|
||
PecContactCreate,
|
||
PecContactImportResult,
|
||
PecContactListResponse,
|
||
PecContactResponse,
|
||
PecContactUpdate,
|
||
)
|
||
from app.services.pec_contact_service import PecContactService
|
||
|
||
router = APIRouter(tags=["Contacts"])
|
||
|
||
|
||
@router.get("/contacts", response_model=PecContactListResponse)
|
||
async def list_contacts(
|
||
current_user: CurrentUser,
|
||
db: DB,
|
||
q: str | None = Query(None, description="Ricerca per email, nome o organizzazione"),
|
||
page: int = Query(1, ge=1),
|
||
page_size: int = Query(50, ge=1, le=200),
|
||
) -> PecContactListResponse:
|
||
"""Elenca i contatti della rubrica PEC del tenant."""
|
||
svc = PecContactService(db)
|
||
items, total = await svc.list_contacts(
|
||
current_user.tenant_id, q=q, page=page, page_size=page_size
|
||
)
|
||
return PecContactListResponse(
|
||
items=[PecContactResponse.model_validate(c) for c in items],
|
||
total=total,
|
||
)
|
||
|
||
|
||
@router.get("/contacts/autocomplete", response_model=list[PecContactResponse])
|
||
async def autocomplete_contacts(
|
||
q: str,
|
||
current_user: CurrentUser,
|
||
db: DB,
|
||
limit: int = Query(10, ge=1, le=20),
|
||
) -> list[PecContactResponse]:
|
||
"""Ricerca rapida contatti per autocomplete nel compose (minimo 2 caratteri)."""
|
||
svc = PecContactService(db)
|
||
items = await svc.search_for_autocomplete(current_user.tenant_id, q=q, limit=limit)
|
||
return [PecContactResponse.model_validate(c) for c in items]
|
||
|
||
|
||
@router.post("/contacts", response_model=PecContactResponse, status_code=status.HTTP_201_CREATED)
|
||
async def create_contact(
|
||
data: PecContactCreate,
|
||
current_user: CurrentUser,
|
||
db: DB,
|
||
) -> PecContactResponse:
|
||
"""Aggiunge un contatto manualmente alla rubrica."""
|
||
svc = PecContactService(db)
|
||
contact = await svc.create_contact(
|
||
current_user.tenant_id, data, created_by=current_user.id
|
||
)
|
||
return PecContactResponse.model_validate(contact)
|
||
|
||
|
||
@router.get("/contacts/{contact_id}", response_model=PecContactResponse)
|
||
async def get_contact(
|
||
contact_id: uuid.UUID,
|
||
current_user: CurrentUser,
|
||
db: DB,
|
||
) -> PecContactResponse:
|
||
"""Restituisce il dettaglio di un contatto."""
|
||
svc = PecContactService(db)
|
||
contact = await svc.get_contact(current_user.tenant_id, contact_id)
|
||
return PecContactResponse.model_validate(contact)
|
||
|
||
|
||
@router.put("/contacts/{contact_id}", response_model=PecContactResponse)
|
||
async def update_contact(
|
||
contact_id: uuid.UUID,
|
||
data: PecContactUpdate,
|
||
current_user: CurrentUser,
|
||
db: DB,
|
||
) -> PecContactResponse:
|
||
"""Aggiorna un contatto della rubrica."""
|
||
svc = PecContactService(db)
|
||
contact = await svc.update_contact(current_user.tenant_id, contact_id, data)
|
||
return PecContactResponse.model_validate(contact)
|
||
|
||
|
||
@router.delete("/contacts/{contact_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||
async def delete_contact(
|
||
contact_id: uuid.UUID,
|
||
current_user: CurrentUser,
|
||
db: DB,
|
||
) -> None:
|
||
"""Elimina un contatto dalla rubrica."""
|
||
svc = PecContactService(db)
|
||
await svc.delete_contact(current_user.tenant_id, contact_id)
|
||
|
||
|
||
@router.post("/contacts/import", response_model=PecContactImportResult)
|
||
async def import_contacts_csv(
|
||
file: UploadFile = File(..., description="File CSV con colonne: email, name, organization"),
|
||
current_user: CurrentUser = ..., # type: ignore
|
||
db: DB = ..., # type: ignore
|
||
) -> PecContactImportResult:
|
||
"""
|
||
Importa contatti dalla rubrica da file CSV.
|
||
|
||
Il CSV deve avere le intestazioni: email, name, organization
|
||
(solo email e' obbligatoria).
|
||
"""
|
||
content = await file.read()
|
||
try:
|
||
csv_text = content.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
csv_text = content.decode("latin-1")
|
||
|
||
svc = PecContactService(db)
|
||
return await svc.import_csv(current_user.tenant_id, csv_text, created_by=current_user.id)
|