diff options
Diffstat (limited to 'generalresearch/managers/thl/profiling')
| -rw-r--r-- | generalresearch/managers/thl/profiling/__init__.py | 0 | ||||
| -rw-r--r-- | generalresearch/managers/thl/profiling/question.py | 157 | ||||
| -rw-r--r-- | generalresearch/managers/thl/profiling/schema.py | 75 | ||||
| -rw-r--r-- | generalresearch/managers/thl/profiling/uqa.py | 211 | ||||
| -rw-r--r-- | generalresearch/managers/thl/profiling/user_upk.py | 343 |
5 files changed, 786 insertions, 0 deletions
diff --git a/generalresearch/managers/thl/profiling/__init__.py b/generalresearch/managers/thl/profiling/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/generalresearch/managers/thl/profiling/__init__.py diff --git a/generalresearch/managers/thl/profiling/question.py b/generalresearch/managers/thl/profiling/question.py new file mode 100644 index 0000000..1ad27ac --- /dev/null +++ b/generalresearch/managers/thl/profiling/question.py @@ -0,0 +1,157 @@ +import random +import threading +from typing import Collection, List, Tuple + +from cachetools import cached, TTLCache +from pydantic import ValidationError + +from generalresearch.decorators import LOG +from generalresearch.managers.base import PostgresManager +from generalresearch.models.thl.profiling.upk_question import ( + UpkQuestion, + UPKImportance, +) + + +class QuestionManager(PostgresManager): + + def get_multi_upk(self, question_ids: Collection[str]) -> List[UpkQuestion]: + query = """ + SELECT data, property_code, explanation_template, explanation_fragment_template + FROM marketplace_question + WHERE id = ANY(%(question_ids)s); + """ + res = self.pg_config.execute_sql_query( + query, {"question_ids": list(question_ids)} + ) + for x in res: + x["data"]["ext_question_id"] = x["property_code"] + x["data"]["explanation_template"] = x["explanation_template"] + x["data"]["explanation_fragment_template"] = x[ + "explanation_fragment_template" + ] + x["data"].pop("categories", None) + return [UpkQuestion.model_validate(x["data"]) for x in res] + + @cached( + cache=TTLCache(maxsize=256, ttl=3600 + random.randint(-900, 900)), + lock=threading.Lock(), + info=True, + ) + def get_questions_ranked( + self, country_iso: str, language_iso: str + ) -> List[UpkQuestion]: + query = """ + SELECT data, property_code, explanation_template, explanation_fragment_template + FROM marketplace_question + WHERE country_iso = %(country_iso)s + AND language_iso = %(language_iso)s + AND property_code NOT LIKE 'gr:%%' + AND property_code NOT LIKE 'g:%%' + AND is_live + """ + res = self.pg_config.execute_sql_query( + query=query, + params={"country_iso": country_iso, "language_iso": language_iso}, + ) + qs: List[UpkQuestion] = [] + for x in res: + x["data"]["ext_question_id"] = x["property_code"] + x["data"]["explanation_template"] = x["explanation_template"] + x["data"]["explanation_fragment_template"] = x[ + "explanation_fragment_template" + ] + x["data"].pop("categories", None) + q = UpkQuestion.model_validate(x["data"]) + if not q.importance: + q.importance = UPKImportance( + task_count=x["data"].get("task_count", 0), + task_score=x["data"].get("task_score", 0), + ) + qs.append(q) + + res = sorted(qs, key=lambda x: x.importance.task_score, reverse=True) + return res + + @cached( + cache=TTLCache(maxsize=256, ttl=3600 + random.randint(-900, 900)), + lock=threading.Lock(), + info=True, + ) + def lookup_by_property( + self, property_code: str, country_iso: str, language_iso: str + ) -> UpkQuestion: + query = f""" + SELECT data, property_code, explanation_template, explanation_fragment_template + FROM marketplace_question + WHERE property_code = %(property_code)s + AND country_iso = %(country_iso)s + AND language_iso = %(language_iso)s + LIMIT 2; + """ + params = { + "property_code": property_code, + "country_iso": country_iso, + "language_iso": language_iso, + } + res = self.pg_config.execute_sql_query(query=query, params=params) + assert len(res) == 1, f"expected 1, got {len(res)} results" + x = res[0] + x["data"]["ext_question_id"] = x["property_code"] + x["data"]["explanation_template"] = x["explanation_template"] + x["data"]["explanation_fragment_template"] = x["explanation_fragment_template"] + x["data"].pop("categories", None) + return UpkQuestion.model_validate(x["data"]) + + def filter_by_property( + self, lookup: Collection[Tuple[str, str, str]] + ) -> List[UpkQuestion]: + """ + lookup is [(property_code, country_iso, language_iso)] + """ + where_str = " OR ".join( + "(property_code = %s AND country_iso = %s AND language_iso = %s)" + for _ in lookup + ) + query = f""" + SELECT data, property_code, explanation_template, explanation_fragment_template + FROM marketplace_question + WHERE {where_str} + """ + flat_params = [item for tup in lookup for item in tup] + res = self.pg_config.execute_sql_query(query, params=flat_params) + for x in res: + x["data"]["ext_question_id"] = x["property_code"] + x["data"]["explanation_template"] = x["explanation_template"] + x["data"]["explanation_fragment_template"] = x[ + "explanation_fragment_template" + ] + x["data"].pop("categories", None) + res2 = [] + for x in res: + try: + res2.append(UpkQuestion.model_validate(x["data"])) + except ValidationError as e: + LOG.warning(e) + return res2 + + def update_question_explanation(self, q: UpkQuestion): + # Assuming the question already exists in the db, and we're updating + # the fields explanation_template and explanation_fragment_template + assert q.id, "q.id must be set" + query = """ + UPDATE marketplace_question + SET explanation_template = %(explanation_template)s, + explanation_fragment_template = %(explanation_fragment_template)s + WHERE id = %(id)s;""" + params = { + "id": q.id, + "explanation_template": q.explanation_template, + "explanation_fragment_template": q.explanation_fragment_template, + } + with self.pg_config.make_connection() as conn: + with conn.cursor() as c: + c.execute(query, params) + assert c.rowcount == 1 + conn.commit() + return None diff --git a/generalresearch/managers/thl/profiling/schema.py b/generalresearch/managers/thl/profiling/schema.py new file mode 100644 index 0000000..581270b --- /dev/null +++ b/generalresearch/managers/thl/profiling/schema.py @@ -0,0 +1,75 @@ +from threading import RLock +from typing import List +from uuid import UUID + +from cachetools import cached, TTLCache + +from generalresearch.managers.base import PostgresManager +from generalresearch.models.thl.profiling.upk_property import ( + UpkProperty, +) + + +class UpkSchemaManager(PostgresManager): + + @cached(cache=TTLCache(maxsize=1, ttl=18 * 60), lock=RLock()) + def get_props_info(self) -> List[UpkProperty]: + query = """ + SELECT + p.id AS property_id, + p.label AS property_label, + p.cardinality, + p.prop_type, + pc.country_iso, + pc.gold_standard, + allowed_items.allowed_items, + cats.categories + FROM marketplace_property p + JOIN marketplace_propertycountry pc + ON p.id = pc.property_id + -- allowed_items: all items for this property + country + LEFT JOIN LATERAL ( + SELECT jsonb_agg( + jsonb_build_object( + 'id', mi.id, + 'label', mi.label, + 'description', mi.description + ) ORDER BY mi.label + ) AS allowed_items + FROM marketplace_propertyitemrange pir + JOIN marketplace_item mi + ON pir.item_id = mi.id + WHERE pir.property_id = p.id + AND pir.country_iso = pc.country_iso + ) allowed_items ON TRUE + + -- categories: all categories for this property + LEFT JOIN LATERAL ( + SELECT + jsonb_agg( + jsonb_build_object( + 'uuid', cat.uuid, + 'label', cat.label, + 'path', cat.path, + 'adwords_vertical_id', cat.adwords_vertical_id + ) + ) AS categories + FROM marketplace_propertycategoryassociation pcat + JOIN marketplace_category cat ON pcat.category_id = cat.id + WHERE pcat.property_id = p.id + ) AS cats ON TRUE; + """ + res = self.pg_config.execute_sql_query(query) + for x in res: + for c in x["categories"]: + c["uuid"] = UUID(c["uuid"]).hex + if x["allowed_items"]: + for c in x["allowed_items"]: + c["id"] = UUID(c["id"]).hex + return [UpkProperty.model_validate(x) for x in res] + + def get_props_info_for_country(self, country_iso: str) -> List[UpkProperty]: + assert country_iso.lower() == country_iso + res = self.get_props_info() + res = [x for x in res if x.country_iso == country_iso].copy() + return res diff --git a/generalresearch/managers/thl/profiling/uqa.py b/generalresearch/managers/thl/profiling/uqa.py new file mode 100644 index 0000000..6800d32 --- /dev/null +++ b/generalresearch/managers/thl/profiling/uqa.py @@ -0,0 +1,211 @@ +import logging +from datetime import datetime, timedelta, timezone +from typing import Collection, List, Optional + +from generalresearch.managers.base import PostgresManagerWithRedis +from generalresearch.models.thl.profiling.user_question_answer import ( + DUMMY_UQA, + UserQuestionAnswer, +) +from generalresearch.models.thl.user import User + +logger = logging.getLogger() + + +class UQAManager(PostgresManagerWithRedis): + + CACHE_PREFIX = "thl-grpc:uqa-cache-v2" + + def redis_key(self, user: User) -> str: + return f"{self.CACHE_PREFIX}:{user.user_id}" + + def redis_lock_key(self, user: User) -> str: + return self.redis_key(user) + ":lock" + + def update_cache( + self, + user: User, + uqas: List[UserQuestionAnswer], + ): + """ + Adds new answers to the redis cache for this user. If the cache + doesn't exist, does a db query to populate the cache. + """ + REDIS = self.redis_client + redis_key = self.redis_key(user) + redis_lock = self.redis_lock_key(user) + json_answers = [uqa.model_dump_json() for uqa in uqas] + with REDIS.lock(redis_lock, timeout=2): + # Append the new answers into the cache. + res = REDIS.rpushx(redis_key, *json_answers) + if res == 0: + # If the cache doesn't exist, we need to make it from scratch + uqas = self.get_from_db(user) + all_json_answers = {uqa.model_dump_json() for uqa in uqas} + # And then make sure thew new answers are in it (b/c we query from the RR) + all_json_answers.update(set(json_answers)) + REDIS.rpush(redis_key, *all_json_answers) + REDIS.expire(redis_key, 3600) + self.clear_user_demographic_cache(user) + return res + + def clear_cache(self, user: User) -> None: + self.redis_client.delete(self.redis_key(user)) + + def recreate_cache(self, user: User) -> Collection[UserQuestionAnswer]: + REDIS = self.redis_client + redis_key = self.redis_key(user) + redis_lock = self.redis_lock_key(user) + with REDIS.lock(redis_lock, timeout=2): + # once we've acquired the lock, we need to check again if someone else just made it + if REDIS.exists(redis_key): + values = REDIS.lrange(redis_key, 0, -1) + uqas = [UserQuestionAnswer.model_validate_json(x) for x in values] + else: + uqas = self.get_from_db(user) + if not uqas: + # We can't set this to an empty list in redis (no + # difference between None and []). + # + # I don't know what is the best thing to do here, so I'm + # gonna push in a "dummy" UQA, just to indicate this has + # been set and we shouldn't query the db anymore unless + # something changes... + REDIS.rpush(redis_key, DUMMY_UQA.model_dump_json()) + else: + REDIS.rpush(redis_key, *[uqa.model_dump_json() for uqa in uqas]) + REDIS.expire(redis_key, 3600 * 3 * 24) + + uqas = self._dedupe_and_clean_uqas(uqas) + self.clear_user_demographic_cache(user) + return uqas + + def _dedupe_and_clean_uqas( + self, + uqas: List[UserQuestionAnswer], + ) -> List[UserQuestionAnswer]: + # Remove anything older than 30 days + uqas = [uqa for uqa in uqas if not uqa.is_stale()] + + # Dedupe, latest answer per question + new_uqas = set() + seen_question_ids = set() + uqas = sorted(uqas, key=lambda x: x.timestamp, reverse=True) + for uqa in uqas: + if uqa.question_id not in seen_question_ids: + seen_question_ids.add(uqa.question_id) + new_uqas.add(uqa) + + return sorted(new_uqas, key=lambda x: x.timestamp, reverse=True) + + def get(self, user: User) -> List[UserQuestionAnswer]: + uqas = self.get_from_cache(user=user) + + if uqas is None: + uqas = self.recreate_cache(user) + return self._dedupe_and_clean_uqas(uqas) + + def get_from_cache(self, user: User) -> Optional[List[UserQuestionAnswer]]: + redis_key = self.redis_key(user) + + # Do the exists check and the list retrieval in a single transaction + with self.redis_client.pipeline() as pipe: + exists = pipe.exists(redis_key) + if exists: + pipe.lrange(redis_key, 0, -1) + result = pipe.execute() + exists = result[0] + values = result[1] + if not exists: + logger.info(f"{redis_key} doesn't exist") + return None + uqas = [UserQuestionAnswer.model_validate_json(x) for x in values] + logger.info(f"{redis_key} exists") + return uqas + + def get_from_db(self, user: User) -> List[UserQuestionAnswer]: + logger.info(f"get_uqa_from_db: {user.user_id}") + # Only store the latest row per question_id. We don't need it multiple times. + since = datetime.now(tz=timezone.utc) - timedelta(days=30) + + # We CAN use the RR, b/c either + # 1) the cache expired and the user hasn't sent an answer recently + # or 2) The user just sent an answer, so we'll make sure it gets put into the results + # after this query runs. + query = f""" + WITH ranked AS ( + SELECT + uqa.*, + ROW_NUMBER() OVER ( + PARTITION BY question_id + ORDER BY created DESC + ) AS rn + FROM marketplace_userquestionanswer uqa + WHERE uqa.user_id = %(user_id)s + AND uqa.created > %(since)s + ) + SELECT + r.question_id::uuid, + r.created::timestamptz AS timestamp, + r.calc_answer::jsonb AS calc_answers, + r.answer::jsonb, + r.user_id, + mq.property_code, + mq.country_iso, + mq.language_iso + FROM ranked r + JOIN marketplace_question mq + ON r.question_id = mq.id + WHERE rn = 1 + ORDER BY r.created; + """ + res = self.pg_config.execute_sql_query( + query=query, + params={"user_id": user.user_id, "since": since}, + ) + uqas = [UserQuestionAnswer.model_validate(x) for x in res] + return uqas + + def clear_user_demographic_cache( + self, + user: User, + ) -> None: + # this will get regenerated by thl-grpc when an offerwall call is made + redis_key = f"thl-grpc:user-demographics:{user.user_id}" + self.redis_client.delete(redis_key) + + return None + + def create( + self, + user: User, + uqas: List[UserQuestionAnswer], + session_id: Optional[str] = None, + ): + for uqa in uqas: + if uqa.user_id is None: + uqa.user_id = user.user_id + else: + assert uqa.user_id == user.user_id + self.create_in_db(uqas=uqas, session_id=session_id) + self.update_cache(user=user, uqas=uqas) + return None + + def create_in_db( + self, uqas: List[UserQuestionAnswer], session_id: Optional[str] = None + ): + values = [uqa.model_dump_mysql(session_id=session_id) for uqa in uqas] + query = """ + INSERT INTO marketplace_userquestionanswer + (created, session_id, answer, question_id, user_id, calc_answer) + VALUES ( + %(created)s, %(session_id)s, %(answer)s, + %(question_id)s, %(user_id)s, %(calc_answer)s + ); + """ + + with self.pg_config.make_connection() as conn: + with conn.cursor() as c: + c.executemany(query=query, params_seq=values) + conn.commit() + return None diff --git a/generalresearch/managers/thl/profiling/user_upk.py b/generalresearch/managers/thl/profiling/user_upk.py new file mode 100644 index 0000000..4449afa --- /dev/null +++ b/generalresearch/managers/thl/profiling/user_upk.py @@ -0,0 +1,343 @@ +import json +from collections import defaultdict +from datetime import timedelta, datetime, timezone +from typing import Dict, Union, Set, List, Collection, Optional, Tuple +from uuid import UUID + +from psycopg import Cursor + +from generalresearch.managers.base import ( + Permission, + PostgresManagerWithRedis, +) +from generalresearch.managers.thl.profiling.schema import UpkSchemaManager +from generalresearch.models.thl.profiling.upk_property import ( + Cardinality, + PropertyType, + UpkProperty, +) +from generalresearch.models.thl.profiling.upk_question_answer import ( + UpkQuestionAnswer, +) +from generalresearch.pg_helper import PostgresConfig +from generalresearch.redis_helper import RedisConfig + + +class UserUpkManager(PostgresManagerWithRedis): + def __init__( + self, + pg_config: PostgresConfig, + redis_config: RedisConfig, + permissions: Collection[Permission] = None, + cache_prefix: Optional[str] = None, + ): + super().__init__( + pg_config=pg_config, + redis_config=redis_config, + permissions=permissions, + cache_prefix=cache_prefix, + ) + self.upk_schema_manager = UpkSchemaManager(pg_config=pg_config) + + def clear_upk_cache(self, user_id: int) -> None: + self.redis_client.delete(f"thl-grpc:user-upk:{user_id}") + return None + + def get_user_upk(self, user_id: int) -> List[UpkQuestionAnswer]: + res = self.redis_client.get(f"thl-grpc:user-upk:{user_id}") + if res: + return [UpkQuestionAnswer.model_validate(x) for x in json.loads(res)] + res = self.get_user_upk_mysql(user_id) + value = json.dumps([x.model_dump(mode="json") for x in res]) + self.redis_client.set(f"thl-grpc:user-upk:{user_id}", value, ex=60 * 60 * 24) + return res + + def get_user_upk_mysql(self, user_id: int) -> List[UpkQuestionAnswer]: + since = datetime.now(tz=timezone.utc) - timedelta(days=89) + + query = """ + SELECT + x.property_id, + mp.label AS property_label, + mi.id AS item_id, + mi.label AS item_label, + mp.prop_type, + mp.cardinality, + x.created, + x.value_num, + x.value_text, + x.country_iso + FROM ( + SELECT + property_id, + value::uuid AS item_id, + NULL::numeric AS value_num, + NULL::text AS value_text, + created, + country_iso + FROM marketplace_userprofileknowledgeitem AS upki + WHERE user_id = %(user_id)s AND created > %(since)s + + UNION ALL + + SELECT + property_id, + NULL::uuid AS item_id, + value::numeric AS value_num, + NULL::text AS value_text, + created, + country_iso + FROM marketplace_userprofileknowledgenumerical + WHERE user_id = %(user_id)s AND created > %(since)s + + UNION ALL + + SELECT + property_id, + NULL::uuid AS item_id, + NULL::numeric AS value_num, + value::text AS value_text, + created, + country_iso + FROM marketplace_userprofileknowledgetext + WHERE user_id = %(user_id)s AND created > %(since)s + ) x + JOIN marketplace_property mp + ON x.property_id = mp.id + LEFT JOIN marketplace_item mi + ON x.item_id = mi.id::uuid; + """ + params = {"user_id": user_id, "since": since} + res = self.pg_config.execute_sql_query(query, params=params) + for x in res: + x["user_id"] = user_id + x["property_id"] = UUID(x["property_id"]).hex + if x["item_id"]: + x["item_id"] = UUID(x["item_id"]).hex + return [UpkQuestionAnswer.model_validate(x) for x in res] + + def get_user_upk_simple( + self, user_id, country_iso="us" + ) -> Dict[str, Union[Set[str], str, float]]: + res = self.get_user_upk(user_id=user_id) + res = [x for x in res if x.country_iso == country_iso] + d: Dict[str, Union[Set[str], str, float]] = defaultdict(set) + for x in res: + if x.cardinality == Cardinality.ZERO_OR_ONE: + d[x.property_label] = x.value + else: + d[x.property_label].add(x.value) + return dict(d) + + def get_age_gender( + self, user_id, country_iso="us" + ) -> Tuple[Optional[int], Optional[str]]: + # Returns an integer year for age, and {'male', 'female', 'other_gender'} + d = self.get_user_upk_simple(user_id, country_iso) + age = d.get("age_in_years") + if age is not None: + age = int(age) + gender = d.get("gender") + return age, gender + + def get_upk_schema(self, country_iso: str) -> List[UpkProperty]: + return self.upk_schema_manager.get_props_info_for_country( + country_iso=country_iso + ) + + def populate_user_upk_from_dict(self, upk_ans_dict): + + country_isos = {x["country_iso"] for x in upk_ans_dict} + assert len(country_isos) == 1 + country_iso = list(country_isos)[0] + for x in upk_ans_dict: + x["pred"] = x["pred"].replace("gr:", "") + x["obj"] = x["obj"].replace("gr:", "") + prop_labels = {x["pred"] for x in upk_ans_dict} + + props = self.get_upk_schema(country_iso=country_iso) + props = [ + x + for x in props + if x.property_label in prop_labels or x.property_id in prop_labels + ] + label_to_prop = {x.property_label: x for x in props} + id_to_prop = {x.property_id: x for x in props} + + for x in upk_ans_dict: + prop = label_to_prop.get(x["pred"]) or id_to_prop[x["pred"]] + x["property_id"] = prop.property_id + x["property_label"] = prop.property_label + x["prop_type"] = prop.prop_type + x["cardinality"] = prop.cardinality + x["created"] = x["timestamp"] + if prop.prop_type == PropertyType.UPK_ITEM: + if x["obj"] in prop.allowed_items_by_id: + x["item_label"] = prop.allowed_items_by_id[x["obj"]].label + x["item_id"] = x["obj"] + else: + x["item_label"] = x["obj"] + x["item_id"] = prop.allowed_items_by_label[x["obj"]].id + elif prop.prop_type == PropertyType.UPK_TEXT: + x["value_text"] = x["obj"] + elif prop.prop_type == PropertyType.UPK_NUMERICAL: + x["value_num"] = x["obj"] + + upk_ans = [UpkQuestionAnswer.model_validate(x) for x in upk_ans_dict] + return upk_ans + + def upsert_user_profile_knowledge(self, c: Cursor, row: UpkQuestionAnswer): + prop_type_table = { + PropertyType.UPK_ITEM: "marketplace_userprofileknowledgeitem", + PropertyType.UPK_NUMERICAL: "marketplace_userprofileknowledgenumerical", + PropertyType.UPK_TEXT: "marketplace_userprofileknowledgetext", + } + prop_type_value = { + PropertyType.UPK_ITEM: "item_id", + PropertyType.UPK_NUMERICAL: "value_num", + PropertyType.UPK_TEXT: "value_text", + } + table = prop_type_table[row.prop_type] + value = prop_type_value[row.prop_type] + args = row.model_dump_mysql() + + c.execute( + f""" + SELECT id FROM {table} + WHERE user_id = %(user_id)s AND property_id = %(property_id)s AND country_iso = %(country_iso)s + LIMIT 1""", + args, + ) + existing = c.fetchone() + + if existing: + c.execute( + f""" + UPDATE {table} + SET value = %({value})s, + created = %(created)s, + question_id = %(question_id)s, + session_id = %(session_id)s + WHERE user_id = %(user_id)s AND + property_id = %(property_id)s AND + country_iso = %(country_iso)s + """, + args, + ) + else: + c.execute( + f""" + INSERT INTO {table} + (property_id, value, created, country_iso, question_id, user_id, session_id) + VALUES (%(property_id)s, %({value})s, %(created)s, %(country_iso)s, + %(question_id)s, %(user_id)s, %(session_id)s) + """, + args, + ) + + def upsert_user_profile_knowledge_multi_item( + self, c: Cursor, row: UpkQuestionAnswer + ): + args = row.model_dump_mysql() + + c.execute( + """ + SELECT id FROM marketplace_userprofileknowledgeitem + WHERE user_id = %(user_id)s AND + property_id = %(property_id)s AND + country_iso = %(country_iso)s AND + value = %(item_id)s + LIMIT 1""", + args, + ) + existing = c.fetchone() + + if existing: + c.execute( + """ + UPDATE marketplace_userprofileknowledgeitem + SET created = %(created)s, + question_id = %(question_id)s, + session_id = %(session_id)s + WHERE user_id = %(user_id)s AND + property_id = %(property_id)s AND + country_iso = %(country_iso)s AND + value = %(item_id)s + """, + args, + ) + else: + c.execute( + """ + INSERT INTO marketplace_userprofileknowledgeitem + (property_id, value, created, country_iso, question_id, user_id, session_id) + VALUES (%(property_id)s, %(item_id)s, %(created)s, %(country_iso)s, + %(question_id)s, %(user_id)s, %(session_id)s) + """, + args, + ) + + def delete_user_profile_knowledge_multi_item( + self, c: Cursor, row: UpkQuestionAnswer + ) -> None: + args = row.model_dump_mysql() + c.execute( + """ + DELETE FROM marketplace_userprofileknowledgeitem + WHERE user_id = %(user_id)s AND + property_id = %(property_id)s AND + country_iso = %(country_iso)s AND + value = %(item_id)s + """, + args, + ) + + return None + + def set_user_upk(self, upk_ans: List[UpkQuestionAnswer]): + user_id = {x.user_id for x in upk_ans} + assert len(user_id) == 1, "only run for 1 user at a time" + user_id = list(user_id)[0] + + curr_upk = self.get_user_upk(user_id=user_id) + curr_upk_simple = self.get_user_upk_simple(user_id=user_id) + + new_upk_simple = defaultdict(set) + delete_items = set() + upk_multi = list() + delete_upk_multi = list() + for x in upk_ans: + # For zero or more (multiple values) We want all values to equal these. + # Might involve deleting values if they exist and are not in upk_ans + if ( + x.cardinality == Cardinality.ZERO_OR_MORE + and x.prop_type != PropertyType.UPK_ITEM + ): + raise ValueError("unsupported") + if ( + x.cardinality == Cardinality.ZERO_OR_MORE + and x.prop_type == PropertyType.UPK_ITEM + ): + new_upk_simple[x.property_label].add(x.item_label) + upk_multi.append(x) + for k, v in new_upk_simple.items(): + prop_delete_labels = curr_upk_simple.get(k, set()) - v + for x in prop_delete_labels: + delete_items.add((k, x)) + if delete_items: + for x in curr_upk: + if (x.property_label, x.item_label) in delete_items: + delete_upk_multi.append(x) + + with self.pg_config.make_connection() as conn: + with conn.cursor() as c: + for x in upk_multi: + self.upsert_user_profile_knowledge_multi_item(c, row=x) + for x in delete_upk_multi: + self.delete_user_profile_knowledge_multi_item(c, row=x) + for x in upk_ans: + if x.cardinality == Cardinality.ZERO_OR_ONE: + # If the cardinality is 0 or 1, we're inserting or updating the answer + self.upsert_user_profile_knowledge(c, x) + conn.commit() + self.clear_upk_cache(user_id=user_id) |