From 854509526a28b788818c37ba9373fe415e974fb7 Mon Sep 17 00:00:00 2001 From: Jacob Brown Date: Mon, 10 Nov 2025 10:30:50 -0700 Subject: [PATCH 1/3] feat: get polymorphic record via function --- services/util.py | 36 ++++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/services/util.py b/services/util.py index cb3d8826c..eb77a3721 100644 --- a/services/util.py +++ b/services/util.py @@ -5,6 +5,7 @@ import httpx from constants import SRID_WGS84 +from db import Base TRANSFORMERS = {} @@ -115,6 +116,41 @@ def get_epqs_elevation_from_point(lon: float, lat: float) -> float | None: return data["value"] +def retrieve_polymorphic_table_record( + target_record: Base, + polymorphic_relationship: str, + polymorphic_type: str, + latest=True, +) -> Base: + """ + Retrieve a record from a polymorphic table. This function assumes that the + parent class has the correct mixin to support retrieval via an attribute. + Parameters: + ---------- + target_record : Base + The parent record from which to retrieve the polymorphic child record. + polymorphic_relationship : str + The name of the relationship attribute on the parent record that corresponds to the polymorphic table. + polymorphic_type : str + The specific type of the polymorphic record to retrieve (e.g., 'Use Status' or 'Monitoring Status' for StatusHistory). + latest : bool, optional + If True, retrieves the latest record based on start_date. Defaults to True. + """ + if polymorphic_relationship == "permissions": + type_field = "permission_type" + elif polymorphic_relationship == "status_history": + type_field = "status_type" + + polymorphic_records = getattr(target_record, polymorphic_relationship) + type_polymorphic_records = [ + r for r in polymorphic_records if getattr(r, type_field) == polymorphic_type + ] + sorted_type_polymorphic_records = sorted( + type_polymorphic_records, key=lambda r: r.start_date, reverse=latest + ) + return sorted_type_polymorphic_records[0] + + if __name__ == "__main__": x = -106.904107 y = 34.068198 From 86ec948dfda5642129280c30b7f60a45b95d0d8d Mon Sep 17 00:00:00 2001 From: Jacob Brown Date: Mon, 10 Nov 2025 15:32:39 -0700 Subject: [PATCH 2/3] refactor: latest record must have null end date --- services/util.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/services/util.py b/services/util.py index eb77a3721..e9e286a6e 100644 --- a/services/util.py +++ b/services/util.py @@ -116,15 +116,16 @@ def get_epqs_elevation_from_point(lon: float, lat: float) -> float | None: return data["value"] -def retrieve_polymorphic_table_record( +def retrieve_latest_polymorphic_table_record( target_record: Base, polymorphic_relationship: str, polymorphic_type: str, - latest=True, ) -> Base: """ - Retrieve a record from a polymorphic table. This function assumes that the - parent class has the correct mixin to support retrieval via an attribute. + Retrieve the latest record from a polymorphic table. This function assumes that the + parent class has the correct mixin to support retrieval via an attribute. This + requires end_date to be None + Parameters: ---------- target_record : Base @@ -143,10 +144,12 @@ def retrieve_polymorphic_table_record( polymorphic_records = getattr(target_record, polymorphic_relationship) type_polymorphic_records = [ - r for r in polymorphic_records if getattr(r, type_field) == polymorphic_type + r + for r in polymorphic_records + if getattr(r, type_field) == polymorphic_type and r.end_date is None ] sorted_type_polymorphic_records = sorted( - type_polymorphic_records, key=lambda r: r.start_date, reverse=latest + type_polymorphic_records, key=lambda r: r.start_date, reverse=True ) return sorted_type_polymorphic_records[0] From 239ba3e05731881fe19c7696ee8505001e59e01b Mon Sep 17 00:00:00 2001 From: Jacob Brown Date: Mon, 10 Nov 2025 15:38:32 -0700 Subject: [PATCH 3/3] refactor: move polymorphic record retrival to tests --- services/util.py | 39 --------------------------------------- tests/__init__.py | 38 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 39 deletions(-) diff --git a/services/util.py b/services/util.py index e9e286a6e..cb3d8826c 100644 --- a/services/util.py +++ b/services/util.py @@ -5,7 +5,6 @@ import httpx from constants import SRID_WGS84 -from db import Base TRANSFORMERS = {} @@ -116,44 +115,6 @@ def get_epqs_elevation_from_point(lon: float, lat: float) -> float | None: return data["value"] -def retrieve_latest_polymorphic_table_record( - target_record: Base, - polymorphic_relationship: str, - polymorphic_type: str, -) -> Base: - """ - Retrieve the latest record from a polymorphic table. This function assumes that the - parent class has the correct mixin to support retrieval via an attribute. This - requires end_date to be None - - Parameters: - ---------- - target_record : Base - The parent record from which to retrieve the polymorphic child record. - polymorphic_relationship : str - The name of the relationship attribute on the parent record that corresponds to the polymorphic table. - polymorphic_type : str - The specific type of the polymorphic record to retrieve (e.g., 'Use Status' or 'Monitoring Status' for StatusHistory). - latest : bool, optional - If True, retrieves the latest record based on start_date. Defaults to True. - """ - if polymorphic_relationship == "permissions": - type_field = "permission_type" - elif polymorphic_relationship == "status_history": - type_field = "status_type" - - polymorphic_records = getattr(target_record, polymorphic_relationship) - type_polymorphic_records = [ - r - for r in polymorphic_records - if getattr(r, type_field) == polymorphic_type and r.end_date is None - ] - sorted_type_polymorphic_records = sorted( - type_polymorphic_records, key=lambda r: r.start_date, reverse=True - ) - return sorted_type_polymorphic_records[0] - - if __name__ == "__main__": x = -106.904107 y = 34.068198 diff --git a/tests/__init__.py b/tests/__init__.py index 1dbba0836..e5937e75d 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -109,4 +109,42 @@ def cleanup_patch_test(model: Base, payload: dict, original_data: Base) -> None: session.commit() +def retrieve_latest_polymorphic_table_record( + target_record: Base, + polymorphic_relationship: str, + polymorphic_type: str, +) -> Base: + """ + Retrieve the latest record from a polymorphic table. This function assumes that the + parent class has the correct mixin to support retrieval via an attribute. This + requires end_date to be None + + Parameters: + ---------- + target_record : Base + The parent record from which to retrieve the polymorphic child record. + polymorphic_relationship : str + The name of the relationship attribute on the parent record that corresponds to the polymorphic table. + polymorphic_type : str + The specific type of the polymorphic record to retrieve (e.g., 'Use Status' or 'Monitoring Status' for StatusHistory). + latest : bool, optional + If True, retrieves the latest record based on start_date. Defaults to True. + """ + if polymorphic_relationship == "permissions": + type_field = "permission_type" + elif polymorphic_relationship == "status_history": + type_field = "status_type" + + polymorphic_records = getattr(target_record, polymorphic_relationship) + type_polymorphic_records = [ + r + for r in polymorphic_records + if getattr(r, type_field) == polymorphic_type and r.end_date is None + ] + sorted_type_polymorphic_records = sorted( + type_polymorphic_records, key=lambda r: r.start_date, reverse=True + ) + return sorted_type_polymorphic_records[0] + + # ============= EOF =============================================