Skip to content

Commit a728492

Browse files
abrookinsclaude
abrookins
and
claude
committed
Refactor datetime conversion exception handling
Replace nested exception blocks with cleaner helper functions to improve maintainability and debuggability. Eliminates broad exception catching that could mask real bugs while preserving datetime conversion functionality. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <[email protected]>
1 parent 589668b commit a728492

File tree

1 file changed

+111
-78
lines changed

1 file changed

+111
-78
lines changed

‎aredis_om/model/model.py‎

Lines changed: 111 additions & 78 deletions
Original file line numberDiff line numberDiff line change
@@ -71,100 +71,133 @@ def convert_datetime_to_timestamp(obj):
7171
return obj
7272

7373

74+
def _extract_base_type(field_type):
75+
"""Extract the base type from Optional or Union types."""
76+
if hasattr(field_type, "__origin__") and field_type.__origin__ is Union:
77+
# For Optional[T] which is Union[T, None], get the non-None type
78+
args = getattr(field_type, "__args__", ())
79+
non_none_types = [
80+
arg for arg in args if getattr(arg, "__name__", None) != "NoneType"
81+
]
82+
if len(non_none_types) == 1:
83+
return non_none_types[0]
84+
return field_type
85+
86+
87+
def _is_datetime_type(field_type):
88+
"""Check if field type is a datetime or date type."""
89+
return field_type in (datetime.datetime, datetime.date)
90+
91+
92+
def _has_model_fields(type_obj):
93+
"""Safely check if a type has model_fields attribute."""
94+
return (
95+
isinstance(type_obj, type)
96+
and hasattr(type_obj, "model_fields")
97+
and type_obj.model_fields
98+
)
99+
100+
101+
def _convert_timestamp_value(value, target_type):
102+
"""Convert a timestamp value to datetime/date with proper error handling."""
103+
if not isinstance(value, (int, float, str)):
104+
return value
105+
106+
try:
107+
if isinstance(value, str):
108+
value = float(value)
109+
110+
# Convert to datetime - preserve original timezone behavior exactly
111+
dt = datetime.datetime.fromtimestamp(value)
112+
113+
# Return date if target is date type
114+
if target_type is datetime.date:
115+
return dt.date()
116+
else:
117+
return dt
118+
119+
except (ValueError, OSError, OverflowError):
120+
# Invalid timestamp, return original value
121+
return value
122+
123+
124+
def _get_list_inner_type(field_type):
125+
"""Extract inner type from List[T] annotation."""
126+
if (
127+
hasattr(field_type, "__origin__")
128+
and field_type.__origin__ in (list, List)
129+
and hasattr(field_type, "__args__")
130+
and field_type.__args__
131+
):
132+
return field_type.__args__[0]
133+
return None
134+
135+
74136
def convert_timestamp_to_datetime(obj, model_fields):
75137
"""Convert Unix timestamps back to datetime objects based on model field types."""
76138
if isinstance(obj, dict):
77139
result = {}
78140
for key, value in obj.items():
79141
if key in model_fields:
80142
field_info = model_fields[key]
81-
field_type = (
82-
field_info.annotation if hasattr(field_info, "annotation") else None
83-
)
143+
field_type = getattr(field_info, "annotation", None)
84144

85-
# Handle Optional types - extract the inner type
86-
if hasattr(field_type, "__origin__") and field_type.__origin__ is Union:
87-
# For Optional[T] which is Union[T, None], get the non-None type
88-
args = getattr(field_type, "__args__", ())
89-
non_none_types = [
90-
arg
91-
for arg in args
92-
if getattr(arg, "__name__", None) != "NoneType"
93-
]
94-
if len(non_none_types) == 1:
95-
field_type = non_none_types[0]
96-
97-
# Handle direct datetime/date fields
98-
if field_type in (datetime.datetime, datetime.date) and isinstance(
99-
value, (int, float, str)
100-
):
101-
try:
102-
if isinstance(value, str):
103-
value = float(value)
104-
# Use fromtimestamp to preserve local timezone behavior
105-
dt = datetime.datetime.fromtimestamp(value)
106-
# If the field is specifically a date, convert to date
107-
if field_type is datetime.date:
108-
result[key] = dt.date()
109-
else:
110-
result[key] = dt
111-
except (ValueError, OSError):
112-
result[key] = value # Keep original value if conversion fails
113-
# Handle nested models - check if it's a RedisModel subclass
114-
elif isinstance(value, dict):
115-
try:
116-
# Check if field_type is a class and subclass of RedisModel
117-
if (
118-
isinstance(field_type, type)
119-
and hasattr(field_type, "model_fields")
120-
and field_type.model_fields
121-
):
122-
result[key] = convert_timestamp_to_datetime(
123-
value, field_type.model_fields
124-
)
125-
else:
126-
result[key] = convert_timestamp_to_datetime(value, {})
127-
except (TypeError, AttributeError):
128-
result[key] = convert_timestamp_to_datetime(value, {})
129-
# Handle lists that might contain nested models
130-
elif isinstance(value, list):
131-
# Try to extract the inner type from List[SomeModel]
132-
inner_type = None
133-
if (
134-
hasattr(field_type, "__origin__")
135-
and field_type.__origin__ in (list, List)
136-
and hasattr(field_type, "__args__")
137-
and field_type.__args__
145+
if field_type:
146+
base_type = _extract_base_type(field_type)
147+
148+
# Handle datetime/date fields
149+
if _is_datetime_type(base_type) and isinstance(
150+
value, (int, float, str)
138151
):
139-
inner_type = field_type.__args__[0]
140-
141-
# Check if the inner type is a nested model
142-
try:
143-
if (
144-
isinstance(inner_type, type)
145-
and hasattr(inner_type, "model_fields")
146-
and inner_type.model_fields
147-
):
148-
result[key] = [
149-
convert_timestamp_to_datetime(
150-
item, inner_type.model_fields
151-
)
152-
for item in value
153-
]
154-
else:
155-
result[key] = convert_timestamp_to_datetime(value, {})
156-
except (TypeError, AttributeError):
152+
result[key] = _convert_timestamp_value(value, base_type)
153+
154+
# Handle nested dictionaries (models)
155+
elif isinstance(value, dict):
156+
nested_fields = (
157+
base_type.model_fields
158+
if _has_model_fields(base_type)
159+
else {}
160+
)
161+
result[key] = convert_timestamp_to_datetime(
162+
value, nested_fields
163+
)
164+
165+
# Handle lists
166+
elif isinstance(value, list):
167+
inner_type = _get_list_inner_type(field_type)
168+
if inner_type and _has_model_fields(inner_type):
169+
result[key] = [
170+
convert_timestamp_to_datetime(
171+
item, inner_type.model_fields
172+
)
173+
for item in value
174+
]
175+
else:
157176
result[key] = convert_timestamp_to_datetime(value, {})
177+
178+
# Handle other types
158179
else:
159-
result[key] = convert_timestamp_to_datetime(value, {})
180+
if isinstance(value, (dict, list)):
181+
result[key] = convert_timestamp_to_datetime(value, {})
182+
else:
183+
result[key] = value
160184
else:
161-
result[key] = convert_timestamp_to_datetime(value, {})
185+
# No field type info, recurse for collections
186+
if isinstance(value, (dict, list)):
187+
result[key] = convert_timestamp_to_datetime(value, {})
188+
else:
189+
result[key] = value
162190
else:
163-
# For keys not in model_fields, still recurse but with empty field info
164-
result[key] = convert_timestamp_to_datetime(value, {})
191+
# Key not in model fields, recurse for collections
192+
if isinstance(value, (dict, list)):
193+
result[key] = convert_timestamp_to_datetime(value, {})
194+
else:
195+
result[key] = value
165196
return result
197+
166198
elif isinstance(obj, list):
167199
return [convert_timestamp_to_datetime(item, model_fields) for item in obj]
200+
168201
else:
169202
return obj
170203

0 commit comments

Comments
 (0)