INNER CODE UNIT · Python
get_cached_data_source
airweave-ai/error-monitoring-agent · backend/sources/__init__.py:110
def get_cached_data_source() -> DataSource:
"""Get cached data source instance."""
global _data_source
if _data_source is None:
_data_source = get_data_source()
return _data_source
def reset_data_source():
"""Reset cached data source (for testing or config changes)."""
global _data_source
_data_source = None
# Export all
__all__ = [
"DataSource",
"RawError",