33 lines
820 B
Python
33 lines
820 B
Python
|
"""
|
||
|
|
Contact helpers for resolving linked customers.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from typing import List, Optional
|
||
|
|
|
||
|
|
from app.core.database import execute_query
|
||
|
|
|
||
|
|
|
||
|
|
def get_contact_customer_ids(contact_id: int) -> List[int]:
|
||
|
|
query = """
|
||
|
|
SELECT customer_id
|
||
|
|
FROM contact_companies
|
||
|
|
WHERE contact_id = %s
|
||
|
|
ORDER BY is_primary DESC, customer_id
|
||
|
|
"""
|
||
|
|
rows = execute_query(query, (contact_id,)) or []
|
||
|
|
return [row["customer_id"] for row in rows]
|
||
|
|
|
||
|
|
|
||
|
|
def get_primary_customer_id(contact_id: int) -> Optional[int]:
|
||
|
|
query = """
|
||
|
|
SELECT customer_id
|
||
|
|
FROM contact_companies
|
||
|
|
WHERE contact_id = %s
|
||
|
|
ORDER BY is_primary DESC, customer_id
|
||
|
|
LIMIT 1
|
||
|
|
"""
|
||
|
|
rows = execute_query(query, (contact_id,)) or []
|
||
|
|
if not rows:
|
||
|
|
return None
|
||
|
|
return rows[0]["customer_id"]
|