A: Add address_service — CRUD with tenant isolation, set_default, migration helper
This commit is contained in:
@@ -0,0 +1,272 @@
|
||||
"""Address service — CRUD with tenant isolation, set_default logic."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select, update
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.audit import log_audit
|
||||
from app.models.address import Address
|
||||
|
||||
|
||||
VALID_ENTITY_TYPES = {"company", "contact"}
|
||||
VALID_ADDRESS_TYPES = {"billing", "shipping", "headquarters", "branch", "private", "other"}
|
||||
|
||||
|
||||
def _address_to_dict(a: Address) -> dict[str, Any]:
|
||||
"""Serialize an Address ORM object to dict."""
|
||||
return {
|
||||
"id": str(a.id),
|
||||
"entity_type": a.entity_type,
|
||||
"entity_id": str(a.entity_id),
|
||||
"label": a.label,
|
||||
"address_type": a.address_type,
|
||||
"street": a.street,
|
||||
"street_number": a.street_number,
|
||||
"city": a.city,
|
||||
"zip": a.zip,
|
||||
"state": a.state,
|
||||
"country": a.country,
|
||||
"is_default": a.is_default,
|
||||
"created_at": a.created_at.isoformat() if a.created_at else None,
|
||||
"updated_at": a.updated_at.isoformat() if a.updated_at else None,
|
||||
}
|
||||
|
||||
|
||||
async def list_addresses(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
entity_type: str,
|
||||
entity_id: uuid.UUID,
|
||||
) -> dict[str, Any]:
|
||||
"""List all addresses for a given entity within a tenant."""
|
||||
q = (
|
||||
select(Address)
|
||||
.where(
|
||||
Address.tenant_id == tenant_id,
|
||||
Address.entity_type == entity_type,
|
||||
Address.entity_id == entity_id,
|
||||
Address.deleted_at.is_(None),
|
||||
)
|
||||
.order_by(Address.is_default.desc(), Address.label.asc())
|
||||
)
|
||||
result = await db.execute(q)
|
||||
addresses = result.scalars().all()
|
||||
return {
|
||||
"items": [_address_to_dict(a) for a in addresses],
|
||||
"total": len(addresses),
|
||||
}
|
||||
|
||||
|
||||
async def create_address(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
user_id: uuid.UUID,
|
||||
data: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""Create a new address. If is_default=True, unset other defaults of same type first."""
|
||||
entity_type = data["entity_type"]
|
||||
address_type = data["address_type"]
|
||||
|
||||
if entity_type not in VALID_ENTITY_TYPES:
|
||||
raise ValueError(f"Invalid entity_type: {entity_type}")
|
||||
if address_type not in VALID_ADDRESS_TYPES:
|
||||
raise ValueError(f"Invalid address_type: {address_type}")
|
||||
|
||||
if data.get("is_default"):
|
||||
await db.execute(
|
||||
update(Address)
|
||||
.where(
|
||||
Address.tenant_id == tenant_id,
|
||||
Address.entity_type == entity_type,
|
||||
Address.entity_id == uuid.UUID(data["entity_id"]),
|
||||
Address.address_type == address_type,
|
||||
Address.is_default.is_(True),
|
||||
Address.deleted_at.is_(None),
|
||||
)
|
||||
.values(is_default=False)
|
||||
)
|
||||
|
||||
address = Address(
|
||||
tenant_id=tenant_id,
|
||||
entity_type=entity_type,
|
||||
entity_id=uuid.UUID(data["entity_id"]),
|
||||
label=data["label"],
|
||||
address_type=address_type,
|
||||
street=data.get("street"),
|
||||
street_number=data.get("street_number"),
|
||||
city=data.get("city"),
|
||||
zip=data.get("zip"),
|
||||
state=data.get("state"),
|
||||
country=data.get("country"),
|
||||
is_default=data.get("is_default", False),
|
||||
)
|
||||
db.add(address)
|
||||
await db.flush()
|
||||
await db.refresh(address)
|
||||
await log_audit(
|
||||
db, tenant_id, user_id, "create", "address", address.id,
|
||||
changes={"label": address.label, "entity_type": entity_type},
|
||||
)
|
||||
return _address_to_dict(address)
|
||||
|
||||
|
||||
async def update_address(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
user_id: uuid.UUID,
|
||||
address_id: uuid.UUID,
|
||||
data: dict[str, Any],
|
||||
) -> dict[str, Any] | None:
|
||||
"""Update an address. If setting is_default=True, unset other defaults of same type first."""
|
||||
q = select(Address).where(
|
||||
Address.id == address_id,
|
||||
Address.tenant_id == tenant_id,
|
||||
Address.deleted_at.is_(None),
|
||||
)
|
||||
result = await db.execute(q)
|
||||
address = result.scalar_one_or_none()
|
||||
if address is None:
|
||||
return None
|
||||
|
||||
if data.get("is_default") is True and not address.is_default:
|
||||
await db.execute(
|
||||
update(Address)
|
||||
.where(
|
||||
Address.tenant_id == tenant_id,
|
||||
Address.entity_type == address.entity_type,
|
||||
Address.entity_id == address.entity_id,
|
||||
Address.address_type == address.address_type,
|
||||
Address.is_default.is_(True),
|
||||
Address.id != address_id,
|
||||
Address.deleted_at.is_(None),
|
||||
)
|
||||
.values(is_default=False)
|
||||
)
|
||||
|
||||
changes: dict[str, Any] = {}
|
||||
for field in ("label", "address_type", "street", "street_number", "city", "zip", "state", "country", "is_default"):
|
||||
if field in data and data[field] is not None:
|
||||
old_val = getattr(address, field)
|
||||
changes[field] = {"old": old_val, "new": data[field]}
|
||||
setattr(address, field, data[field])
|
||||
|
||||
await db.flush()
|
||||
await db.refresh(address)
|
||||
await log_audit(db, tenant_id, user_id, "update", "address", address_id, changes=changes)
|
||||
return _address_to_dict(address)
|
||||
|
||||
|
||||
async def delete_address(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
user_id: uuid.UUID,
|
||||
address_id: uuid.UUID,
|
||||
) -> bool:
|
||||
"""Soft-delete an address."""
|
||||
q = select(Address).where(
|
||||
Address.id == address_id,
|
||||
Address.tenant_id == tenant_id,
|
||||
Address.deleted_at.is_(None),
|
||||
)
|
||||
result = await db.execute(q)
|
||||
address = result.scalar_one_or_none()
|
||||
if address is None:
|
||||
return False
|
||||
|
||||
address.deleted_at = datetime.now(UTC)
|
||||
await db.flush()
|
||||
await log_audit(
|
||||
db, tenant_id, user_id, "delete", "address", address_id,
|
||||
changes={"label": address.label},
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
async def migrate_existing_addresses(db: AsyncSession) -> int:
|
||||
"""Migrate existing address fields from companies/contacts to Address records.
|
||||
Called during migration 0013. Returns count of created addresses.
|
||||
"""
|
||||
from app.models.company import Company
|
||||
from app.models.contact import Contact
|
||||
|
||||
count = 0
|
||||
|
||||
# Migrate company addresses
|
||||
companies_q = select(Company).where(
|
||||
Company.address_street.is_not(None),
|
||||
Company.deleted_at.is_(None),
|
||||
)
|
||||
companies_result = await db.execute(companies_q)
|
||||
for company in companies_result.scalars():
|
||||
existing = await db.execute(
|
||||
select(Address).where(
|
||||
Address.tenant_id == company.tenant_id,
|
||||
Address.entity_type == "company",
|
||||
Address.entity_id == company.id,
|
||||
Address.address_type == "headquarters",
|
||||
Address.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
if existing.scalar_one_or_none() is not None:
|
||||
continue
|
||||
|
||||
addr = Address(
|
||||
tenant_id=company.tenant_id,
|
||||
entity_type="company",
|
||||
entity_id=company.id,
|
||||
label="Hauptsitz",
|
||||
address_type="headquarters",
|
||||
street=company.address_street,
|
||||
city=company.address_city,
|
||||
zip=company.address_zip,
|
||||
country=company.address_country,
|
||||
state=company.address_state,
|
||||
is_default=True,
|
||||
)
|
||||
db.add(addr)
|
||||
count += 1
|
||||
|
||||
# Migrate contact addresses
|
||||
contacts_q = select(Contact).where(
|
||||
Contact.address_street.is_not(None),
|
||||
Contact.deleted_at.is_(None),
|
||||
)
|
||||
contacts_result = await db.execute(contacts_q)
|
||||
for contact in contacts_result.scalars():
|
||||
existing = await db.execute(
|
||||
select(Address).where(
|
||||
Address.tenant_id == contact.tenant_id,
|
||||
Address.entity_type == "contact",
|
||||
Address.entity_id == contact.id,
|
||||
Address.address_type == "private",
|
||||
Address.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
if existing.scalar_one_or_none() is not None:
|
||||
continue
|
||||
|
||||
addr = Address(
|
||||
tenant_id=contact.tenant_id,
|
||||
entity_type="contact",
|
||||
entity_id=contact.id,
|
||||
label="Privat",
|
||||
address_type="private",
|
||||
street=contact.address_street,
|
||||
city=contact.address_city,
|
||||
zip=contact.address_zip,
|
||||
country=contact.address_country,
|
||||
state=contact.address_state,
|
||||
is_default=True,
|
||||
)
|
||||
db.add(addr)
|
||||
count += 1
|
||||
|
||||
if count > 0:
|
||||
await db.flush()
|
||||
|
||||
return count
|
||||
Reference in New Issue
Block a user