Json Exporter

Author: MuddyHope

examples/async_example.py

@@ -1,8 +1,10 @@
 import asyncio
-import sys
 import os
+import sys
+
 from pyquerytracker import TrackQuery
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
 
 # Example 1: Basic async function

examples/core/async_usage.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+
 from pyquerytracker import TrackQuery, configure
 
 # Configure logging to see the output

examples/exporter/csv_exporter_1.py

@@ -2,7 +2,7 @@
 import time
 
 from pyquerytracker import TrackQuery
-from pyquerytracker.config import configure, ExportType
+from pyquerytracker.config import ExportType, configure
 
 os.makedirs("logs-csv", exist_ok=True)
 
@@ -22,17 +22,13 @@ def process_data(x, y):
 
 @TrackQuery()
 def failing_task():
-    time.sleep(0.03)
+    time.sleep(0.03)  # not slow
     raise RuntimeError("This failed intentionally.")
 
 
-# Run
-print("Result:", process_data(5, 7))
-print("Result:", process_data(5, 7))
-print("Result:", process_data(5, 7))
-print("Result:", process_data(5, 7))
-print("Result:", process_data(5, 7))
+# Run tasks
 print("Result:", process_data(5, 7))
+print("Result:", process_data(1, 2))
 
 try:
     failing_task()

examples/exporter/json_exporter.py

@@ -0,0 +1,36 @@
+import os
+import time
+
+from pyquerytracker import TrackQuery
+from pyquerytracker.config import ExportType, configure
+
+os.makedirs("logs-json", exist_ok=True)
+
+configure(
+    slow_log_threshold_ms=50.0,
+    slow_log_level=20,  # INFO
+    export_type=ExportType.JSON,
+    export_path="logs-json/query_logs.json",
+)
+
+
+@TrackQuery()
+def process_data(x, y):
+    time.sleep(0.08)  # triggers slow log
+    return x + y
+
+
+@TrackQuery()
+def failing_task():
+    time.sleep(0.03)  # not slow
+    raise RuntimeError("This failed intentionally.")
+
+
+# Run tasks
+print("Result:", process_data(5, 7))
+print("Result:", process_data(1, 2))
+
+try:
+    failing_task()
+except Exception:
+    pass

pyquerytracker/exporter/base.py

@@ -1,4 +1,5 @@
 from abc import ABC, abstractmethod
+
 from pyquerytracker.config import Config
 
 

pyquerytracker/exporter/csv_exporter.py

@@ -3,8 +3,8 @@
 import os
 from threading import Lock
 
-from pyquerytracker.utils.logger import QueryLogger
 from pyquerytracker.exporter.base import Exporter
+from pyquerytracker.utils.logger import QueryLogger
 
 logger = QueryLogger.get_logger()
 

pyquerytracker/exporter/json_exporter.py

@@ -0,0 +1,47 @@
+import atexit
+import json
+import os
+from threading import Lock
+
+from pyquerytracker.exporter.base import Exporter
+from pyquerytracker.utils.logger import QueryLogger
+
+logger = QueryLogger.get_logger()
+
+
+class JsonExporter(Exporter):
+    def __init__(self, config):
+        super().__init__(config)
+        self.config = config
+        self._lock = Lock()
+        self._buffer = []
+        atexit.register(self.flush)
+
+    def append(self, data: dict):
+        with self._lock:
+            self._buffer.append(data)
+
+    def flush(self):
+        with self._lock:
+            if not self._buffer:
+                return
+
+            os.makedirs(os.path.dirname(self.config.export_path), exist_ok=True)
+
+            existing_data = []
+            if os.path.exists(self.config.export_path):
+                try:
+                    with open(self.config.export_path, "r", encoding="utf-8") as f:
+                        existing_data = json.load(f)
+                        if not isinstance(existing_data, list):
+                            existing_data = []
+                except json.JSONDecodeError:
+                    pass
+
+            existing_data.extend(self._buffer)
+
+            with open(self.config.export_path, "w", encoding="utf-8") as f:
+                json.dump(existing_data, f, indent=2)
+
+            logger.info(f"Flushed {len(self._buffer)} logs to JSON")
+            self._buffer.clear()

pyquerytracker/exporter/manager.py

@@ -1,18 +1,25 @@
 from pyquerytracker.config import Config, ExportType
-from pyquerytracker.exporter.csv_exporter import CsvExporter
 from pyquerytracker.exporter.base import Exporter
+from pyquerytracker.exporter.csv_exporter import CsvExporter
+from pyquerytracker.exporter.json_exporter import JsonExporter
 
 
 class ExporterManager:
     _exporter: Exporter = None
 
-    @staticmethod
-    def create_exporter(config: Config) -> Exporter:
-        if config.export_type == ExportType.CSV:
-            exporter = CsvExporter(config)
-            ExporterManager.set(exporter)
-            return exporter
-        raise ValueError("Unsupported export type")
+    _exporter_classes = {
+        ExportType.CSV: CsvExporter,
+        ExportType.JSON: JsonExporter,
+    }
+
+    @classmethod
+    def create_exporter(cls, config: Config) -> Exporter:
+        exporter_cls = cls._exporter_classes.get(config.export_type)
+        if not exporter_cls:
+            raise ValueError(f"Unsupported export type: {config.export_type}")
+        exporter = exporter_cls(config)
+        cls.set(exporter)
+        return exporter
 
     @staticmethod
     def set(exporter: Exporter):

tests/exporter/test_json_exporter.py

@@ -0,0 +1,156 @@
+import json
+import os
+import subprocess
+import tempfile
+
+
+def run_test_in_subprocess(script: str, export_path: str):
+    subprocess.run(["python3", "-c", script], check=True)
+    assert os.path.exists(export_path)
+    with open(export_path) as f:
+        return json.load(f)
+
+
+def test_successful_function_logs():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        export_path = os.path.join(tmpdir, "log.json")
+
+        script = f"""
+import time
+from pyquerytracker import TrackQuery
+from pyquerytracker.config import configure, ExportType
+
+configure(
+    slow_log_threshold_ms=0.0,
+    slow_log_level=10,
+    export_type=ExportType.JSON,
+    export_path=r"{export_path}",
+)
+
+@TrackQuery()
+def foo(x, y):
+    return x + y
+
+foo(1, 2)
+"""
+
+        logs = run_test_in_subprocess(script, export_path)
+
+        assert isinstance(logs, list)
+        functions_logged = {log.get("function_name") for log in logs}
+        assert "foo" in functions_logged
+        assert any(log.get("event") == "success" for log in logs) or any(
+            log.get("event") == "slow_execution" for log in logs
+        )
+
+
+def test_error_function_logs():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        export_path = os.path.join(tmpdir, "log.json")
+
+        script = f"""
+import time
+from pyquerytracker import TrackQuery
+from pyquerytracker.config import configure, ExportType
+
+configure(
+    slow_log_threshold_ms=0.0,
+    slow_log_level=10,
+    export_type=ExportType.JSON,
+    export_path=r"{export_path}",
+)
+
+@TrackQuery()
+def bar():
+    raise RuntimeError("error")
+
+try:
+    bar()
+except RuntimeError:
+    pass
+"""
+
+        logs = run_test_in_subprocess(script, export_path)
+
+        assert isinstance(logs, list)
+        functions_logged = {log.get("function_name") for log in logs}
+        assert "bar" in functions_logged
+        assert any(log.get("event") == "error" for log in logs)
+
+
+def test_slow_function_logs():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        export_path = os.path.join(tmpdir, "log.json")
+
+        script = f"""
+import time
+from pyquerytracker import TrackQuery
+from pyquerytracker.config import configure, ExportType
+
+# Set threshold low so the sleep triggers slow log
+configure(
+    slow_log_threshold_ms=5.0,
+    slow_log_level=10,
+    export_type=ExportType.JSON,
+    export_path=r"{export_path}",
+)
+
+@TrackQuery()
+def slow_func():
+    time.sleep(0.01)
+    return "done"
+
+slow_func()
+"""
+
+        logs = run_test_in_subprocess(script, export_path)
+
+        assert isinstance(logs, list)
+        functions_logged = {log.get("function_name") for log in logs}
+        assert "slow_func" in functions_logged
+        assert any(log.get("event") == "slow_execution" for log in logs)
+
+
+def test_multiple_calls_logged():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        export_path = os.path.join(tmpdir, "log.json")
+
+        script = f"""
+import time
+from pyquerytracker import TrackQuery
+from pyquerytracker.config import configure, ExportType
+
+configure(
+    slow_log_threshold_ms=0.0,
+    slow_log_level=10,
+    export_type=ExportType.JSON,
+    export_path=r"{export_path}",
+)
+
+@TrackQuery()
+def a():
+    return 1
+
+@TrackQuery()
+def b():
+    raise ValueError("fail")
+
+a()
+try:
+    b()
+except Exception:
+    pass
+a()
+"""
+
+        logs = run_test_in_subprocess(script, export_path)
+
+        assert isinstance(logs, list)
+        functions_logged = [log.get("function_name") for log in logs]
+
+        # Because of overwrite behavior, logs might only contain last flush,
+        # but often logs include multiple entries. So test at least one a, b occurrence:
+        assert "a" in functions_logged or "b" in functions_logged
+        # Check for presence of success and error events in the batch
+        events = {log.get("event") for log in logs}
+        assert "error" in events or "success" in events or "slow_execution" in events

tests/test_async_core.py

@@ -1,6 +1,8 @@
 import asyncio
 import logging
+
 import pytest
+
 from pyquerytracker import TrackQuery, configure
 
 # Mark all tests in this file as asyncio