"""
Sous-client pour les endpoints /v1/customers.
Classes:
- ``henrri_connect.customers.synchro.SyncCustomersClient``
Accès synchrone aux endpoints clients.
Notes:
- Utiliser de préférence l'objet ``henrri_connect.SyncHenrriClient`` pour acceder aux endpoints.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, overload, Optional
from ..models import Address, Contact, Customer, PagedListResponse, CustomerRequest
from ..models.base import CustomerType
from ..utils import clean
if TYPE_CHECKING:
from ..connect import SyncHenrriClient
BASE_CUSTOMERS = "/v1/customers"
[docs]
class SyncCustomersClient:
"""
Accès synchrone aux endpoints clients.
Arguments
- ``client`` (SyncHenrriClient) : Client HTTP.
Methodes
- ``list_customers`` : Liste les clients avec pagination et filtres optionnels.
- ``add`` : Crée un nouveau client.
- ``get_best_sales`` : Récupère les meilleurs clients.
- ``get_last_used`` : Récupère le dernier client utilisé.
- ``get`` : Récupère un client par son identifiant.
- ``modify`` : Met à jour un client existant.
- ``delete`` : Supprime un client.
- ``get_address`` : Récupère l'adresse d'un client.
- ``list_contacts`` : Liste les contacts d'un client avec pagination et filtres optionnels.
- ``add_contact`` : Crée un nouveau contact.
- ``get_contact`` : Récupère un contact d'un client.
- ``modify_contact`` : Met à jour un contact existant.
- ``delete_contact`` : Supprime un contact.
"""
def __init__(self, client: SyncHenrriClient) -> None:
self._c = client
@overload
def list_customers(
self,
*,
request: CustomerRequest,
with_selected_fields: bool = True,
with_totals: bool = False,
only_current_page: bool = True
) -> PagedListResponse[Customer]:...
@overload
def list_customers(
self,
*,
request: CustomerRequest,
with_selected_fields: bool = False,
with_totals: None = None,
only_current_page: None = None,
):...
[docs]
def list_customers(
self,
*,
request: CustomerRequest,
with_selected_fields: bool = True,
with_totals: Optional[bool] = False,
only_current_page: Optional[bool] = True
):
"""
Liste les clients avec pagination et filtres optionnels.
Arguments
- ``request`` (CustomerRequest) : Paramètres de recherche.
- ``with_selected_fields`` (bool) : Si True, lance une requête de recherche avancée.
- ``with_totals`` (bool) : Si True, renvoie les totaux.
- ``only_current_page`` (bool) : Si True, renvoie uniquement les clients de la page
actuelle.
Returns
- ``PagedListResponse[Customer]`` : Liste paginée de clients.
"""
params = clean(request.model_dump(by_alias=True))
if with_selected_fields:
if with_totals and only_current_page:
params["with_totals"] = with_totals
params["only_current_page"] = only_current_page
resp = self._c.request(
"GET",
f"{BASE_CUSTOMERS}/with-selected-fields",
params=params
)
else:
raise ValueError(
"with_totals and only_current_page must be True if with_selected_fields is True"
)
else:
resp = self._c.request("GET", BASE_CUSTOMERS, params=params)
return PagedListResponse[Customer].model_validate(resp.json())
[docs]
def add(self, customer: Customer) -> Customer:
"""
Crée un nouveau client.
Arguments
- ``customer`` (Customer) : Client à créer.
Returns
- ``Customer`` : Client créé.
"""
resp = self._c.request(
"POST",
BASE_CUSTOMERS,
json=customer.model_dump(by_alias=True, exclude_unset=True, exclude_none=True),
)
return Customer.model_validate(resp.json())
[docs]
def get_best_sales(
self,
*,
year: int
) -> PagedListResponse[Customer]:
"""
Récupère les meilleurs clients.
Arguments
- ``year`` (int) : Année concerne (minimum 2000, maximum 2100).
Returns
- ``PagedListResponse[Customer]`` : Liste de clients.
"""
params = {
"year": year
}
resp = self._c.request("GET", f"{BASE_CUSTOMERS}/best-sales", params=params)
return PagedListResponse[Customer].model_validate(resp.json())
[docs]
def get_last_used(self, customer_type: CustomerType, limit: int) -> Customer:
"""
Récupère les derniers clients utilisés.
Arguments
- ``customer_type`` (CustomerType) : Type de client.
- ``limit`` (int) : Nombre de clients maximum.
Returns
- ``Customer`` : Derniers clients utilisés.
"""
params = {
"Types": customer_type,
"Limit": limit,
}
resp = self._c.request("GET", f"{BASE_CUSTOMERS}/last-used", params=params)
return Customer.model_validate(resp.json())
[docs]
def get(self, customer_id: int) -> Customer:
"""
Récupère un client par son identifiant.
Arguments
- ``customer_id`` (int) : Identifiant du client.
Returns
- ``Customer`` : Client.
"""
resp = self._c.request("GET", f"{BASE_CUSTOMERS}/{customer_id}")
return Customer.model_validate(resp.json())
[docs]
def modify(self, customer_id: int, customer: Customer) -> Customer:
"""
Met à jour un client existant.
Arguments
- ``customer_id`` (int) : Identifiant du client.
- ``customer`` (Customer) : Client à mettre à jour.
Returns
- ``Customer`` : Client mis à jour.
"""
resp = self._c.request(
"PUT",
f"{BASE_CUSTOMERS}/{customer_id}",
json=customer.model_dump(by_alias=True, exclude_unset=True, exclude_none=True),
)
return Customer.model_validate(resp.json())
[docs]
def delete(self, customer_id: int) -> None:
"""
Supprime un client.
Arguments
- ``customer_id`` (int) : Identifiant du client.
Returns
- ``None``
"""
self._c.request("DELETE", f"{BASE_CUSTOMERS}/{customer_id}")
[docs]
def get_address(self, customer_id: int) -> Address:
"""
Récupère l'adresse d'un client.
Arguments
- ``customer_id`` (int) : Identifiant du client.
Returns
- ``Address`` : Adresse du client.
"""
resp = self._c.request("GET", f"{BASE_CUSTOMERS}/{customer_id}/address")
return Address.model_validate(resp.json())