feat: implement V3 Uniform Lookup System with hierarchical overrides and site-based whitelisting

This commit is contained in:
Scott Idem
2026-02-20 14:48:50 -05:00
parent 2b2a2bc00f
commit 6bfbff309a
8 changed files with 390 additions and 51 deletions

View File

@@ -0,0 +1,98 @@
import logging
from typing import List, Optional
from sqlalchemy import text
from app.lib_sql_core import engine
from app.lib_general_v3 import AccountContext
log = logging.getLogger(__name__)
def get_lookup_list_v3(
lu_type: str,
account_ctx: AccountContext,
for_type: Optional[str] = None,
for_id: Optional[int] = None,
include_disabled: bool = False,
whitelist: Optional[List[str]] = None
) -> List[dict]:
"""
Retrieves a ranked, deduplicated list of lookup records.
Priority: Object Override > Account Override > Global Default.
Supports an optional whitelist (List of 'group' strings).
"""
table_name = f"v_lu_v3_{lu_type}"
# We use ROW_NUMBER() to handle the hierarchy
# 1. Object specific (matching for_type and for_id)
# 2. Account specific (matching account_id)
# 3. Global (account_id IS NULL)
# Whitelist logic: If a whitelist is provided, we only want records where
# the group is in that list.
sql = f"""
SELECT * FROM (
SELECT *,
ROW_NUMBER() OVER (
PARTITION BY `group`
ORDER BY
(for_type = :for_type AND for_id = :for_id) DESC,
(account_id = :account_id) DESC,
created_on DESC
) as rank_priority
FROM `{table_name}`
WHERE ((for_type = :for_type AND for_id = :for_id)
OR account_id = :account_id
OR account_id IS NULL)
"""
if whitelist:
sql += " AND `group` IN :whitelist"
sql += f"""
) AS ranked
WHERE rank_priority = 1
"""
if not include_disabled:
sql += " AND enable = 1"
sql += " ORDER BY sort ASC, name ASC"
params = {
"account_id": account_ctx.account_id,
"for_type": for_type,
"for_id": for_id,
"whitelist": tuple(whitelist) if whitelist else None
}
try:
with engine.connect() as conn:
result = conn.execute(text(sql), params)
return [dict(row._mapping) for row in result]
except Exception as e:
log.error(f"Error in get_lookup_list_v3: {e}")
return []
def resolve_lookup_v3(
lu_type: str,
query: str,
account_ctx: AccountContext,
identity_fields: List[str]
) -> Optional[dict]:
"""
Resolves a query string to a single lookup record by scanning multiple identity fields.
Returns the highest-priority match.
"""
# Simple implementation: get the full ranked list and find first match in identity fields
# For performance with large tables (like timezones), we might want a specific SQL query
full_list = get_lookup_list_v3(lu_type, account_ctx)
query_clean = query.strip().lower()
for item in full_list:
for field in identity_fields:
val = item.get(field)
if val and str(val).lower() == query_clean:
return item
return None