From f73457ef725772670969cdd25ce8878b2af77266 Mon Sep 17 00:00:00 2001 From: HectorSin Date: Sat, 10 Jan 2026 20:24:00 +0900 Subject: [PATCH 1/5] refactor: add type hints for user_id and visualization server args Signed-off-by: HectorSin --- cognee/shared/utils.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/cognee/shared/utils.py b/cognee/shared/utils.py index 7c76cfa59..17501a512 100644 --- a/cognee/shared/utils.py +++ b/cognee/shared/utils.py @@ -8,7 +8,8 @@ import http.server import socketserver from threading import Thread import pathlib -from uuid import uuid4, uuid5, NAMESPACE_OID +from typing import Union +from uuid import uuid4, uuid5, NAMESPACE_OID, UUID from cognee.base_config import get_base_config from cognee.shared.logging_utils import get_logger @@ -78,7 +79,7 @@ def _sanitize_nested_properties(obj, property_names: list[str]): return obj -def send_telemetry(event_name: str, user_id, additional_properties: dict = {}): +def send_telemetry(event_name: str, user_id: Union[str, UUID], additional_properties: dict = {}): if os.getenv("TELEMETRY_DISABLED"): return @@ -138,7 +139,9 @@ def embed_logo(p, layout_scale, logo_alpha, position): def start_visualization_server( - host="0.0.0.0", port=8001, handler_class=http.server.SimpleHTTPRequestHandler + host: str = "0.0.0.0", + port: int = 8001, + handler_class=http.server.SimpleHTTPRequestHandler, ): """ Spin up a simple HTTP server in a background thread to serve files. From ebf2aaaa5c4f049bf5e9af893912d7053f175a9c Mon Sep 17 00:00:00 2001 From: HectorSin Date: Sat, 10 Jan 2026 20:58:08 +0900 Subject: [PATCH 2/5] refactor: add type hint for handler_class Signed-off-by: HectorSin --- cognee/shared/utils.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/cognee/shared/utils.py b/cognee/shared/utils.py index 17501a512..460a10bfb 100644 --- a/cognee/shared/utils.py +++ b/cognee/shared/utils.py @@ -141,7 +141,9 @@ def embed_logo(p, layout_scale, logo_alpha, position): def start_visualization_server( host: str = "0.0.0.0", port: int = 8001, - handler_class=http.server.SimpleHTTPRequestHandler, + handler_class: type[ + http.server.SimpleHTTPRequestHandler + ] = http.server.SimpleHTTPRequestHandler, ): """ Spin up a simple HTTP server in a background thread to serve files. From 46c12cc0ee7dda7b379b52f34a3f84f6607b3c6f Mon Sep 17 00:00:00 2001 From: HectorSin Date: Sat, 10 Jan 2026 21:13:10 +0900 Subject: [PATCH 3/5] refactor: resolve remaining ANN001 errors in utils.py Signed-off-by: HectorSin --- cognee/shared/utils.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/cognee/shared/utils.py b/cognee/shared/utils.py index 460a10bfb..7eb036ef1 100644 --- a/cognee/shared/utils.py +++ b/cognee/shared/utils.py @@ -8,7 +8,7 @@ import http.server import socketserver from threading import Thread import pathlib -from typing import Union +from typing import Union, Any, Dict, List from uuid import uuid4, uuid5, NAMESPACE_OID, UUID from cognee.base_config import get_base_config @@ -59,7 +59,7 @@ def get_anonymous_id(): return anonymous_id -def _sanitize_nested_properties(obj, property_names: list[str]): +def _sanitize_nested_properties(obj: Union[Dict, List, Any], property_names: list[str]): """ Recursively replaces any property whose key matches one of `property_names` (e.g., ['url', 'path']) in a nested dict or list with a uuid5 hash @@ -109,7 +109,7 @@ def send_telemetry(event_name: str, user_id: Union[str, UUID], additional_proper print(f"Error sending telemetry through proxy: {response.status_code}") -def embed_logo(p, layout_scale, logo_alpha, position): +def embed_logo(p: Any, layout_scale: float, logo_alpha: float, position: str): """ Embed a logo into the graph visualization as a watermark. """ From da5660b7169c32a744430b06201c282a338dcf76 Mon Sep 17 00:00:00 2001 From: HectorSin Date: Sat, 10 Jan 2026 21:21:04 +0900 Subject: [PATCH 4/5] refactor: fix mutable default argument in send_telemetry Signed-off-by: HectorSin --- cognee/shared/utils.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/cognee/shared/utils.py b/cognee/shared/utils.py index 7eb036ef1..ce9fb2169 100644 --- a/cognee/shared/utils.py +++ b/cognee/shared/utils.py @@ -80,6 +80,8 @@ def _sanitize_nested_properties(obj: Union[Dict, List, Any], property_names: lis def send_telemetry(event_name: str, user_id: Union[str, UUID], additional_properties: dict = {}): + if additional_properties is None: + additional_properties = {} if os.getenv("TELEMETRY_DISABLED"): return From 4189cda89518c779d329b7cd05c9050533b6a40e Mon Sep 17 00:00:00 2001 From: HectorSin Date: Sat, 10 Jan 2026 21:25:25 +0900 Subject: [PATCH 5/5] refactor: simplify type hint and add return type for sanitize function Signed-off-by: HectorSin --- cognee/shared/utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cognee/shared/utils.py b/cognee/shared/utils.py index ce9fb2169..3b6e7ba14 100644 --- a/cognee/shared/utils.py +++ b/cognee/shared/utils.py @@ -59,7 +59,7 @@ def get_anonymous_id(): return anonymous_id -def _sanitize_nested_properties(obj: Union[Dict, List, Any], property_names: list[str]): +def _sanitize_nested_properties(obj: Any, property_names: list[str]) -> Any: """ Recursively replaces any property whose key matches one of `property_names` (e.g., ['url', 'path']) in a nested dict or list with a uuid5 hash