From 7c7966dbd40a743af1af52e5ed6d0af99aa4755a Mon Sep 17 00:00:00 2001
From: Harry Lees <harry.lees@gmail.com>
Date: Fri, 20 Dec 2024 14:32:35 +0000
Subject: [PATCH] Add type annotations to `kombu/utils/encoding.py`

The initial annotations provided aim to annotate the current implementation opposed
to potentially refactoring to narrow down the original intention of the provided
implementations.
---
 kombu/utils/encoding.py | 48 ++++++++++++++++++++++++++++++-----------
 1 file changed, 35 insertions(+), 13 deletions(-)

diff --git a/kombu/utils/encoding.py b/kombu/utils/encoding.py
index 42bf2ce9..e15ac9cf 100644
--- a/kombu/utils/encoding.py
+++ b/kombu/utils/encoding.py
@@ -9,69 +9,84 @@ from __future__ import annotations
 
 import sys
 import traceback
+from typing import Any, Union, Literal, overload, TypeVar
+
+
+T = TypeVar("T")
 
 #: safe_str takes encoding from this file by default.
 #: :func:`set_default_encoding_file` can used to set the
 #: default output file.
-default_encoding_file = None
+default_encoding_file: Union[str, None] = None
 
 
-def set_default_encoding_file(file):
+def set_default_encoding_file(file: str) -> None:
     """Set file used to get codec information."""
     global default_encoding_file
     default_encoding_file = file
 
 
-def get_default_encoding_file():
+def get_default_encoding_file() -> str | None:
     """Get file used to get codec information."""
     return default_encoding_file
 
 
 if sys.platform.startswith('java'):  # pragma: no cover
 
-    def default_encoding(file=None):
+    def default_encoding(file: object = None) -> Literal['utf-8']:
         """Get default encoding."""
         return 'utf-8'
 else:
 
-    def default_encoding(file=None):
+    def default_encoding(file: object = None) -> str:
         """Get default encoding."""
         file = file or get_default_encoding_file()
         return getattr(file, 'encoding', None) or sys.getfilesystemencoding()
 
 
-def str_to_bytes(s):
+@overload
+def str_to_bytes(s: str) -> bytes: ...
+@overload
+def str_to_bytes(s: T) -> T: ...
+def str_to_bytes(s: Any) -> Any:
     """Convert str to bytes."""
     if isinstance(s, str):
         return s.encode()
     return s
 
 
-def bytes_to_str(s):
+@overload
+def bytes_to_str(s: bytes) -> str: ...
+@overload
+def bytes_to_str(s: T) -> T: ...
+def bytes_to_str(s: Any) -> Any:
     """Convert bytes to str."""
     if isinstance(s, bytes):
         return s.decode(errors='replace')
     return s
 
 
-def from_utf8(s, *args, **kwargs):
+def from_utf8(s: str, *args: Any, **kwargs: Any) -> str:
     """Get str from utf-8 encoding."""
     return s
 
 
-def ensure_bytes(s):
+def ensure_bytes(s: Union[str, bytes]) -> bytes:
     """Ensure s is bytes, not str."""
     if not isinstance(s, bytes):
         return str_to_bytes(s)
     return s
 
 
-def default_encode(obj):
+def default_encode(obj: T) -> T:
     """Encode using default encoding."""
     return obj
 
 
-def safe_str(s, errors='replace'):
+def safe_str(
+    s: object,
+    errors: str = 'replace',
+) -> str:
     """Safe form of str(), void of unicode errors."""
     s = bytes_to_str(s)
     if not isinstance(s, (str, bytes)):
@@ -79,7 +94,11 @@ def safe_str(s, errors='replace'):
     return _safe_str(s, errors)
 
 
-def _safe_str(s, errors='replace', file=None):
+def _safe_str(
+    s: object,
+    errors: str = 'replace',
+    file: Any = None
+) -> str:
     if isinstance(s, str):
         return s
     try:
@@ -89,7 +108,10 @@ def _safe_str(s, errors='replace', file=None):
             type(s), exc, '\n'.join(traceback.format_stack()))
 
 
-def safe_repr(o, errors='replace'):
+def safe_repr(
+    o: object,
+    errors: str = 'replace',
+) -> str:
     """Safe form of repr, void of Unicode errors."""
     try:
         return repr(o)