Add address handling for ClientProvider in business catalog creation

- Introduced a new helper function `_ensure_client_provider_address` to manage the assignment of addresses for ClientProvider instances.
- Refactored the `create_business_catalogs` function to utilize the new helper, ensuring that existing addresses are reused if available, preventing duplicate entries.
- This change enhances the integrity of address data during business catalog creation by ensuring proper address assignment based on existing records.
This commit is contained in:
2026-03-24 12:42:19 -05:00
parent b1a74d8e30
commit c812f5214c

View File

@@ -90,6 +90,38 @@ def ensure_tenant_company(db: Session, tenant_id: int = 1, company_id: int = 1)
return company
def _ensure_client_provider_address(
db: Session,
client: ClientProvider,
*,
tenant_id: int,
company_id: int,
country: str,
) -> None:
"""
Garantiza que el cliente tenga fila de dirección. Si el ORM no tiene `address`
cargada pero en BD ya existe (datos previos en CI / DB compartida), reutiliza
esa fila en lugar de INSERT duplicado (pkey id = client_id).
"""
if client.address is not None:
return
existing = (
db.query(ClientProviderAddress)
.filter(ClientProviderAddress.client_id == client.id)
.first()
)
if existing is not None:
client.address = existing
return
client.address = ClientProviderAddress(
id=client.id,
client_id=client.id,
tenant_id=tenant_id,
company_id=company_id,
country=country,
)
def create_business_catalogs(db: Session, tenant_id: int, company_id: int) -> dict:
provider = (
db.query(ClientProvider)
@@ -110,14 +142,13 @@ def create_business_catalogs(db: Session, tenant_id: int, company_id: int) -> di
)
db.add(provider)
db.flush()
if provider.address is None:
provider.address = ClientProviderAddress(
id=provider.id,
client_id=provider.id,
tenant_id=tenant_id,
company_id=company_id,
country="MEX",
)
_ensure_client_provider_address(
db,
provider,
tenant_id=tenant_id,
company_id=company_id,
country="MEX",
)
sold_to = (
db.query(ClientProvider)
@@ -138,14 +169,13 @@ def create_business_catalogs(db: Session, tenant_id: int, company_id: int) -> di
)
db.add(sold_to)
db.flush()
if sold_to.address is None:
sold_to.address = ClientProviderAddress(
id=sold_to.id,
client_id=sold_to.id,
tenant_id=tenant_id,
company_id=company_id,
country="USA",
)
_ensure_client_provider_address(
db,
sold_to,
tenant_id=tenant_id,
company_id=company_id,
country="USA",
)
shipped_to = (
db.query(ClientProvider)
@@ -166,14 +196,13 @@ def create_business_catalogs(db: Session, tenant_id: int, company_id: int) -> di
)
db.add(shipped_to)
db.flush()
if shipped_to.address is None:
shipped_to.address = ClientProviderAddress(
id=shipped_to.id,
client_id=shipped_to.id,
tenant_id=tenant_id,
company_id=company_id,
country="USA",
)
_ensure_client_provider_address(
db,
shipped_to,
tenant_id=tenant_id,
company_id=company_id,
country="USA",
)
broker = (
db.query(CustomsBroker)