A: Remove address fields from company_service — dict, create, update, export

This commit is contained in:
2026-07-04 01:42:10 +00:00
parent 7473efbb02
commit 1708c4d94a
+5 -29
View File
@@ -16,9 +16,6 @@ from app.models.company import Company
from app.models.contact import CompanyContact, Contact from app.models.contact import CompanyContact, Contact
ADDRESS_FIELDS = ("address_street", "address_city", "address_zip", "address_country", "address_state")
def _company_to_dict(c: Company, include_contacts: bool = False) -> dict[str, Any]: def _company_to_dict(c: Company, include_contacts: bool = False) -> dict[str, Any]:
"""Serialize a Company ORM object to dict.""" """Serialize a Company ORM object to dict."""
data: dict[str, Any] = { data: dict[str, Any] = {
@@ -30,11 +27,6 @@ def _company_to_dict(c: Company, include_contacts: bool = False) -> dict[str, An
"email": c.email, "email": c.email,
"website": c.website, "website": c.website,
"description": c.description, "description": c.description,
"address_street": c.address_street,
"address_city": c.address_city,
"address_zip": c.address_zip,
"address_country": c.address_country,
"address_state": c.address_state,
"created_at": c.created_at.isoformat() if c.created_at else None, "created_at": c.created_at.isoformat() if c.created_at else None,
"updated_at": c.updated_at.isoformat() if c.updated_at else None, "updated_at": c.updated_at.isoformat() if c.updated_at else None,
} }
@@ -160,11 +152,6 @@ async def create_company(
email=data.get("email"), email=data.get("email"),
website=data.get("website"), website=data.get("website"),
description=data.get("description"), description=data.get("description"),
address_street=data.get("address_street"),
address_city=data.get("address_city"),
address_zip=data.get("address_zip"),
address_country=data.get("address_country"),
address_state=data.get("address_state"),
created_by=user_id, created_by=user_id,
updated_by=user_id, updated_by=user_id,
) )
@@ -205,7 +192,7 @@ async def update_company(
all_fields = ( all_fields = (
"name", "account_number", "industry", "phone", "email", "name", "account_number", "industry", "phone", "email",
"website", "description", "website", "description",
) + ADDRESS_FIELDS )
for field in all_fields: for field in all_fields:
if field in data and data[field] is not None: if field in data and data[field] is not None:
old_val = getattr(company, field) old_val = getattr(company, field)
@@ -250,12 +237,7 @@ async def soft_delete_company(
await db.flush() await db.flush()
await log_audit( await log_audit(
db, db, tenant_id, user_id, "delete", "company", company_id,
tenant_id,
user_id,
"delete",
"company",
company_id,
changes={"name": company.name, "cascade": cascade}, changes={"name": company.name, "cascade": cascade},
) )
return True return True
@@ -391,15 +373,12 @@ async def export_companies_csv(
output = io.StringIO() output = io.StringIO()
writer = csv.writer(output) writer = csv.writer(output)
writer.writerow( writer.writerow(
["id", "name", "account_number", "industry", "phone", "email", "website", "description", ["id", "name", "account_number", "industry", "phone", "email", "website", "description"]
"address_street", "address_city", "address_zip", "address_country", "address_state"]
) )
for c in companies: for c in companies:
writer.writerow( writer.writerow(
[str(c.id), c.name, c.account_number or "", c.industry or "", [str(c.id), c.name, c.account_number or "", c.industry or "",
c.phone or "", c.email or "", c.website or "", c.description or "", c.phone or "", c.email or "", c.website or "", c.description or ""]
c.address_street or "", c.address_city or "", c.address_zip or "",
c.address_country or "", c.address_state or ""]
) )
return output.getvalue() return output.getvalue()
@@ -438,15 +417,12 @@ async def export_companies_xlsx(
ws.title = "Companies" ws.title = "Companies"
headers = [ headers = [
"id", "name", "account_number", "industry", "phone", "email", "website", "description", "id", "name", "account_number", "industry", "phone", "email", "website", "description",
"address_street", "address_city", "address_zip", "address_country", "address_state",
] ]
ws.append(headers) ws.append(headers)
for c in companies: for c in companies:
ws.append( ws.append(
[str(c.id), c.name, c.account_number or "", c.industry or "", [str(c.id), c.name, c.account_number or "", c.industry or "",
c.phone or "", c.email or "", c.website or "", c.description or "", c.phone or "", c.email or "", c.website or "", c.description or ""]
c.address_street or "", c.address_city or "", c.address_zip or "",
c.address_country or "", c.address_state or ""]
) )
buf = io.BytesIO() buf = io.BytesIO()