my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	import json
#2	import logging
#3	import httpx
#4
#5	from mem0.exceptions import (
#6	NetworkError,
#7	create_exception_from_response,
#8	)
#9
#10	logger = logging.getLogger(__name__)
#11
#12
#13	class APIError(Exception):
#14	"""Exception raised for errors in the API.
#15
#16	Deprecated: Use specific exception classes from mem0.exceptions instead.
#17	This class is maintained for backward compatibility.
#18	"""
#19
#20	pass
#21
#22
#23	def api_error_handler(func):
#24	"""Decorator to handle API errors consistently.
#25
#26	This decorator catches HTTP and request errors and converts them to
#27	appropriate structured exception classes with detailed error information.
#28
#29	The decorator analyzes HTTP status codes and response content to create
#30	the most specific exception type with helpful error messages, suggestions,
#31	and debug information.
#32	"""
#33	from functools import wraps
#34
#35	@wraps(func)
#36	def wrapper(args, *kwargs):
#37	try:
#38	return func(args, *kwargs)
#39	except httpx.HTTPStatusError as e:
#40	logger.error(f"HTTP error occurred: {e}")
#41
#42	# Extract error details from response
#43	response_text = ""
#44	error_details = {}
#45	debug_info = {
#46	"status_code": e.response.status_code,
#47	"url": str(e.request.url),
#48	"method": e.request.method,
#49	}
#50
#51	try:
#52	response_text = e.response.text
#53	# Try to parse JSON response for additional error details
#54	if e.response.headers.get("content-type", "").startswith("application/json"):
#55	error_data = json.loads(response_text)
#56	if isinstance(error_data, dict):
#57	error_details = error_data
#58	response_text = error_data.get("detail", response_text)
#59	except (json.JSONDecodeError, AttributeError):
#60	# Fallback to plain text response
#61	pass
#62
#63	# Add rate limit information if available
#64	if e.response.status_code == 429:
#65	retry_after = e.response.headers.get("Retry-After")
#66	if retry_after:
#67	try:
#68	debug_info["retry_after"] = int(retry_after)
#69	except ValueError:
#70	pass
#71
#72	# Add rate limit headers if available
#73	for header in ["X-RateLimit-Limit", "X-RateLimit-Remaining", "X-RateLimit-Reset"]:
#74	value = e.response.headers.get(header)
#75	if value:
#76	debug_info[header.lower().replace("-", "_")] = value
#77
#78	# Create specific exception based on status code
#79	exception = create_exception_from_response(
#80	status_code=e.response.status_code,
#81	response_text=response_text,
#82	details=error_details,
#83	debug_info=debug_info,
#84	)
#85
#86	raise exception
#87
#88	except httpx.RequestError as e:
#89	logger.error(f"Request error occurred: {e}")
#90
#91	# Determine the appropriate exception type based on error type
#92	if isinstance(e, httpx.TimeoutException):
#93	raise NetworkError(
#94	message=f"Request timed out: {str(e)}",
#95	error_code="NET_TIMEOUT",
#96	suggestion="Please check your internet connection and try again",
#97	debug_info={"error_type": "timeout", "original_error": str(e)},
#98	)
#99	elif isinstance(e, httpx.ConnectError):
#100	raise NetworkError(
#101	message=f"Connection failed: {str(e)}",
#102	error_code="NET_CONNECT",
#103	suggestion="Please check your internet connection and try again",
#104	debug_info={"error_type": "connection", "original_error": str(e)},
#105	)
#106	else:
#107	# Generic network error for other request errors
#108	raise NetworkError(
#109	message=f"Network request failed: {str(e)}",
#110	error_code="NET_GENERIC",
#111	suggestion="Please check your internet connection and try again",
#112	debug_info={"error_type": "request", "original_error": str(e)},
#113	)
#114
#115	return wrapper
#116

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public