128 lines
4.8 KiB
Python
128 lines
4.8 KiB
Python
"""Contact API endpoints."""
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from pydantic import BaseModel, ConfigDict, EmailStr
|
|
|
|
from app.api.deps import get_contact_service, get_organization_context
|
|
from app.models.contact import ContactCreate, ContactRead
|
|
from app.services.contact_service import (
|
|
ContactDeletionError,
|
|
ContactForbiddenError,
|
|
ContactListFilters,
|
|
ContactNotFoundError,
|
|
ContactOrganizationError,
|
|
ContactService,
|
|
ContactUpdateData,
|
|
)
|
|
from app.services.organization_service import OrganizationContext
|
|
|
|
|
|
class ContactCreatePayload(BaseModel):
|
|
name: str
|
|
email: EmailStr | None = None
|
|
phone: str | None = None
|
|
owner_id: int | None = None
|
|
|
|
def to_domain(self, *, organization_id: int, fallback_owner: int) -> ContactCreate:
|
|
return ContactCreate(
|
|
organization_id=organization_id,
|
|
owner_id=self.owner_id or fallback_owner,
|
|
name=self.name,
|
|
email=self.email,
|
|
phone=self.phone,
|
|
)
|
|
|
|
|
|
class ContactUpdatePayload(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
name: str | None = None
|
|
email: EmailStr | None = None
|
|
phone: str | None = None
|
|
|
|
def to_update_data(self) -> ContactUpdateData:
|
|
dump = self.model_dump(exclude_unset=True)
|
|
return ContactUpdateData(
|
|
name=dump.get("name"),
|
|
email=dump.get("email"),
|
|
phone=dump.get("phone"),
|
|
)
|
|
|
|
|
|
router = APIRouter(prefix="/contacts", tags=["contacts"])
|
|
|
|
|
|
@router.get("/", response_model=list[ContactRead])
|
|
async def list_contacts(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
search: str | None = Query(default=None, min_length=1),
|
|
owner_id: int | None = None,
|
|
context: OrganizationContext = Depends(get_organization_context),
|
|
service: ContactService = Depends(get_contact_service),
|
|
) -> list[ContactRead]:
|
|
filters = ContactListFilters(
|
|
page=page,
|
|
page_size=page_size,
|
|
search=search,
|
|
owner_id=owner_id,
|
|
)
|
|
try:
|
|
contacts = await service.list_contacts(filters=filters, context=context)
|
|
except ContactForbiddenError as exc:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
|
|
return [ContactRead.model_validate(contact) for contact in contacts]
|
|
|
|
|
|
@router.post("/", response_model=ContactRead, status_code=status.HTTP_201_CREATED)
|
|
async def create_contact(
|
|
payload: ContactCreatePayload,
|
|
context: OrganizationContext = Depends(get_organization_context),
|
|
service: ContactService = Depends(get_contact_service),
|
|
) -> ContactRead:
|
|
data = payload.to_domain(organization_id=context.organization_id, fallback_owner=context.user_id)
|
|
try:
|
|
contact = await service.create_contact(data, context=context)
|
|
except ContactForbiddenError as exc:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
|
|
except ContactOrganizationError as exc:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
|
|
return ContactRead.model_validate(contact)
|
|
|
|
|
|
@router.patch("/{contact_id}", response_model=ContactRead)
|
|
async def update_contact(
|
|
contact_id: int,
|
|
payload: ContactUpdatePayload,
|
|
context: OrganizationContext = Depends(get_organization_context),
|
|
service: ContactService = Depends(get_contact_service),
|
|
) -> ContactRead:
|
|
try:
|
|
contact = await service.get_contact(contact_id, context=context)
|
|
updated = await service.update_contact(contact, payload.to_update_data(), context=context)
|
|
except ContactNotFoundError as exc:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
|
|
except ContactForbiddenError as exc:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
|
|
except ContactOrganizationError as exc:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
|
|
return ContactRead.model_validate(updated)
|
|
|
|
|
|
@router.delete("/{contact_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_contact(
|
|
contact_id: int,
|
|
context: OrganizationContext = Depends(get_organization_context),
|
|
service: ContactService = Depends(get_contact_service),
|
|
) -> None:
|
|
try:
|
|
contact = await service.get_contact(contact_id, context=context)
|
|
await service.delete_contact(contact, context=context)
|
|
except ContactNotFoundError as exc:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
|
|
except ContactForbiddenError as exc:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
|
|
except ContactDeletionError as exc:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
|