mypy fixes

Author: guenp

build.py

@@ -6,7 +6,7 @@
 
 
 class CustomBuildHook(BuildHookInterface):
-    def initialize(self, version, build_data):
+    def initialize(self, version, build_data):  # noqa: ARG002
         build_data["ext_modules"] = [
             Extension(
                 "telemetric._stats_wrapper",

examples/test_scipy.py

@@ -21,9 +21,9 @@
 # test()
 
 # How many calls (errors, and how often something was odd with the args)
-print("counts: ", stats.norm.pdf._get_counts())
+print("counts: ", stats.norm.pdf._get_counts())  # noqa: T201
 
 # Detailed statistics for each parameter.  How often was it passed
 # and how often were "a" and 3 passed (or something equal to them,
 # I do not consider types right now, although one could)?
-print("param stats: ", stats.norm.pdf._get_param_stats())
+print("param stats: ", stats.norm.pdf._get_param_stats())  # noqa: T201

examples/test_span.py

@@ -8,19 +8,19 @@
 
 @span
 def foo(hello="world", delay=1):
-    print(hello)
+    print(hello)  # noqa: T201
     sleep(delay)
 
 
 @span
 def bar(spam="eggs", delay=1):
-    print(spam)
+    print(spam)  # noqa: T201
     sleep(delay)
 
 
 @span
 def baz(apple="orange", delay=1):
-    print(apple)
+    print(apple)  # noqa: T201
     sleep(delay)
 
 

examples/test_stats_uploader.py

@@ -34,42 +34,42 @@ def another_func(x=1, y=True):
 
 
 # Print current statistics
-print("=== Current Statistics ===")
-print("test_func counts:", test_func._get_counts())
-print("test_func param stats:", test_func._get_param_stats())
-print()
-print("another_func counts:", another_func._get_counts())
-print("another_func param stats:", another_func._get_param_stats())
-print()
+print("=== Current Statistics ===")  # noqa: T201
+print("test_func counts:", test_func._get_counts())  # noqa: T201
+print("test_func param stats:", test_func._get_param_stats())  # noqa: T201
+print()  # noqa: T201
+print("another_func counts:", another_func._get_counts())  # noqa: T201
+print("another_func param stats:", another_func._get_param_stats())  # noqa: T201
+print()  # noqa: T201
 
 
 # Test the stats uploader (this will try to upload to GA4)
-print("=== Testing Stats Uploader ===")
+print("=== Testing Stats Uploader ===")  # noqa: T201
 
 # Use a dummy proxy URL for testing (won't actually send data unless configured)
 uploader = StatsUploader(
     proxy_url="https://analytics-proxy-production-665e.up.railway.app"
 )
 
-print("Analytics client enabled:", uploader.analytics.enabled)
+print("Analytics client enabled:", uploader.analytics.enabled)  # noqa: T201
 
 # Test uploading individual function stats
-print("Uploading test_func stats...")
+print("Uploading test_func stats...")  # noqa: T201
 result1 = uploader.upload_function_stats(test_func, package_name="test_package")
-print("Upload result:", result1)
+print("Upload result:", result1)  # noqa: T201
 
-print("Uploading another_func stats...")
+print("Uploading another_func stats...")  # noqa: T201
 result2 = uploader.upload_function_stats(another_func, package_name="test_package")
-print("Upload result:", result2)
+print("Upload result:", result2)  # noqa: T201
 
 # Test uploading all stats
-print("\nUploading all stats...")
+print("\nUploading all stats...")  # noqa: T201
 summary = uploader.upload_all_stats(package_name="test_package")
-print("Upload summary:", summary)
+print("Upload summary:", summary)  # noqa: T201
 
 # Test custom stats upload
-print("\nUploading custom stats...")
+print("\nUploading custom stats...")  # noqa: T201
 custom_result = uploader.upload_custom_stats(
     "custom_metric", {"metric_name": "test_metric", "value": 42, "category": "testing"}
 )
-print("Custom upload result:", custom_result)
+print("Custom upload result:", custom_result)  # noqa: T201

examples/test_statswrapper.py

@@ -12,9 +12,9 @@ def test_func(a, b=None, c=None):
 test_func(1, 3)
 
 # How many calls (errors, and how often something was odd with the args)
-print("counts: ", test_func._get_counts())
+print("counts: ", test_func._get_counts())  # noqa: T201
 
 # Detailed statistics for each parameter.  How often was it passed
 # and how often were "a" and 3 passed (or something equal to them,
 # I do not consider types right now, although one could)?
-print("param stats: ", test_func._get_param_stats())
+print("param stats: ", test_func._get_param_stats())  # noqa: T201

src/telemetric/console.py

@@ -2,18 +2,24 @@
 
 import os
 
-from opentelemetry import trace
-from opentelemetry.sdk.resources import Resource
-from opentelemetry.sdk.trace import TracerProvider
-from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
+from opentelemetry import trace  # type: ignore[import-not-found]
+from opentelemetry.sdk.resources import Resource  # type: ignore[import-not-found]
+from opentelemetry.sdk.trace import TracerProvider  # type: ignore[import-not-found]
+from opentelemetry.sdk.trace.export import (  # type: ignore[import-not-found]
+    BatchSpanProcessor,
+    ConsoleSpanExporter,
+)
 
 __all__ = ["setup_console"]
 
 
-def setup_console(service_name: str | None = None):
+def setup_console(service_name: str | None = None) -> None:
     if service_name is None:
-        attributes = os.environ.get("OTEL_RESOURCE_ATTRIBUTES")
-        attributes = dict(k.split("=") for k in attributes.split(","))
+        attributes_str = os.environ.get("OTEL_RESOURCE_ATTRIBUTES")
+        if attributes_str:
+            attributes = dict(k.split("=") for k in attributes_str.split(","))
+        else:
+            attributes = {}
     else:
         attributes = {"service.name": service_name}
 

src/telemetric/entrypoint.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
-from opentelemetry.instrumentation.auto_instrumentation import initialize
+from opentelemetry.instrumentation.auto_instrumentation import (  # type: ignore[import-not-found]
+    initialize,
+)
 
 from telemetric import install
 from telemetric.console import setup_console

src/telemetric/ga4/analytics.py

@@ -7,7 +7,7 @@
 import platform
 import uuid
 
-import requests
+import requests  # type: ignore[import-untyped]
 
 _log = logging.getLogger()
 
@@ -24,19 +24,20 @@ class AnalyticsClient:
     })
     """
 
-    def __init__(self, proxy_url):
+    def __init__(self, proxy_url: str) -> None:
         self.proxy_url = proxy_url
         self.client_id = str(uuid.uuid4())
         self.enabled = not self._is_disabled()
 
-    def _is_disabled(self):
-        if os.environ.get("DO_NOT_TRACK", "").lower() in ("1", "true"):
-            return True
-        if os.environ.get("CI", "").lower() in ("1", "true"):
-            return True
-        return False
+    def _is_disabled(self) -> bool:
+        return os.environ.get("DO_NOT_TRACK", "").lower() in (
+            "1",
+            "true",
+        ) or os.environ.get("CI", "").lower() in ("1", "true")
 
-    def track_event(self, event_name, params=None):
+    def track_event(
+        self, event_name: str, params: dict[str, str] | None = None
+    ) -> None:
         if not self.enabled:
             return
 
@@ -60,7 +61,7 @@ def track_event(self, event_name, params=None):
         _log.debug(f"POST request to: {url}")
         _log.debug(payload)
 
-        try:
+        try:  # noqa: SIM105
             requests.post(f"{self.proxy_url}/track", json=payload, timeout=2)
         except Exception:
             pass

src/telemetric/ga4/ga4_proxy.py

@@ -6,10 +6,10 @@
 
 import os
 
-import requests
-from fastapi import FastAPI, Request
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import JSONResponse
+import requests  # type: ignore[import-untyped]
+from fastapi import FastAPI, Request  # type: ignore[import-not-found]
+from fastapi.middleware.cors import CORSMiddleware  # type: ignore[import-not-found]
+from fastapi.responses import JSONResponse  # type: ignore[import-not-found]
 
 app = FastAPI()
 
@@ -28,7 +28,7 @@
 
 
 @app.get("/debug")
-async def debug():
+async def debug():  # type: ignore[no-untyped-def]
     """Debug endpoint to check if env vars are loaded"""
     return {
         "GA4_MEASUREMENT_ID_set": bool(GA4_MEASUREMENT_ID),
@@ -40,13 +40,13 @@ async def debug():
 
 
 @app.get("/")
-async def root():
+async def root():  # type: ignore[no-untyped-def]
     """Health check endpoint"""
     return {"status": "ok", "service": "GA4 Analytics Proxy"}
 
 
-@app.post("/track")
-async def track_event(request: Request):
+@app.post("/track")  # type: ignore[misc]
+async def track_event(request: Request):  # type: ignore[no-untyped-def]
     """
     Proxy endpoint that forwards events to GA4
 
@@ -104,7 +104,7 @@ async def track_event(request: Request):
 
 
 if __name__ == "__main__":
-    import uvicorn
+    import uvicorn  # type: ignore[import-not-found]
 
-    port = int(os.environ.get("PORT", 8000))
+    port = int(os.environ.get("PORT", "8000"))
     uvicorn.run(app, host="0.0.0.0", port=port)

src/telemetric/ga4/stats_uploader.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Any, Dict, Optional
+from typing import Any
 
 from telemetric.ga4.analytics import AnalyticsClient
 from telemetric.statswrapper import _wrapped
@@ -26,8 +26,8 @@ def __init__(self, proxy_url: str):
         """
         self.analytics = AnalyticsClient(proxy_url)
 
-    def upload_function_stats(
-        self, wrapped_func, package_name: Optional[str] = None
+    def upload_function_stats(  # type: ignore[no-untyped-def]
+        self, wrapped_func, package_name: str | None = None
     ) -> bool:
         """
         Upload statistics for a single wrapped function to GA4.
@@ -98,8 +98,8 @@ def upload_function_stats(
         return True
 
     def upload_all_stats(
-        self, package_name: Optional[str] = None, skip_uncalled: bool = True
-    ) -> Dict[str, Any]:
+        self, package_name: str | None = None, skip_uncalled: bool = True
+    ) -> dict[str, Any]:
         """
         Upload statistics for all wrapped functions to GA4.
 
@@ -152,7 +152,7 @@ def upload_all_stats(
             "status": "completed",
         }
 
-    def upload_custom_stats(self, event_name: str, stats_data: Dict[str, Any]) -> bool:
+    def upload_custom_stats(self, event_name: str, stats_data: dict[str, Any]) -> bool:
         """
         Upload custom statistics data to GA4.