Fix utils, update .gitignore:

Add try/except for importing fastapi utils.

Fix imports in all __init__.py files.

Add new operations to msgpack_utils. Add serialize/deserialize simple bytestring.

Fix diskcache always creating a cache by adding get_cache() function and a default cache conf dict.

Author: redjax

.gitignore

@@ -1,5 +1,5 @@
 ## Temporarily ignore .ipynb
-*.ipynb
+# *.ipynb
 
 # Byte-compiled / optimized / DLL files
 __pycache__/
@@ -141,6 +141,8 @@ venv.bak/
 !*.*.*example*
 !*.*.*example*.*
 
+.serialize
+
 # Spyder project settings
 .spyderproject
 .spyproject

red_utils/__init__.py

@@ -4,3 +4,9 @@
 from . import httpx_utils
 from . import loguru_utils
 from . import msgpack_utils
+
+## Only import fastapi utils if fastapi and uvicorn are installed
+try:
+    from . import fastapi_utils
+except:
+    pass

red_utils/diskcache_utils/__init__.py

@@ -1,13 +1,29 @@
 from __future__ import annotations
 
+from typing import Any
+
 from .constants import (
     default_cache_dir,
     default_timeout_dict,
     valid_key_types,
     valid_tag_types,
     valid_val_types,
 )
-from .operations import convert_to_seconds
+
+from .operations import (
+    convert_to_seconds,
+    get_cache,
+    check_cache,
+    get_cache_size,
+    delete_val,
+    get_val,
+    set_expire,
+    set_val,
+    cache_tag_index,
+    check_exists,
+    clear_cache,
+)
+
 from .validators import (
     validate_cache,
     validate_expire,
@@ -19,12 +35,10 @@
     validate_val,
 )
 
-from diskcache import Cache
-
 ## Define a default cache object
-default_cache: Cache = Cache(
-    directory=default_cache_dir,
-    timeout=convert_to_seconds(
+default_cache_conf: dict[str, Any] = {
+    "directory": default_cache_dir,
+    "timeout": convert_to_seconds(
         unit=default_timeout_dict["unit"], amount=default_timeout_dict["amount"]
     ),
-)
+}

red_utils/diskcache_utils/operations.py

@@ -30,6 +30,7 @@
 
 from diskcache import Cache
 
+
 def convert_to_seconds(unit: str = None, amount: int = None) -> int:
     ## Allowed strings for conversion
     valid_time_units: list[int] = ["seconds", "hours", "minutes", "days", "weeks"]
@@ -68,8 +69,32 @@ def convert_to_seconds(unit: str = None, amount: int = None) -> int:
     return _amount
 
 
+# def get_cache(
+#     cache_dir: str = default_cache_dir,
+#     index: bool = True,
+# ) -> diskcache.core.Cache:
+#     """Prepare and return a diskcache.Cache object."""
+#     if not cache_dir:
+#         raise ValueError("Missing cache directory")
+
+#     if not isinstance(cache_dir, Union[str, Path]):
+#         raise TypeError(f"cache_dir must be of type str or Path, not {type(cache_dir)}")
+
+#     try:
+#         _cache: diskcache.core.Cache = Cache(directory=cache_dir)
+
+#         if index:
+#             cache_tag_index(cache=_cache)
+
+#         return _cache
+
+#     except Exception as exc:
+#         raise Exception(f"Unhandled exception creating cache. Details: {exc}")
+
+
 def get_cache(
     cache_dir: str = default_cache_dir,
+    cache_conf: dict = None,
     index: bool = True,
 ) -> diskcache.core.Cache:
     """Prepare and return a diskcache.Cache object."""
@@ -79,8 +104,17 @@ def get_cache(
     if not isinstance(cache_dir, Union[str, Path]):
         raise TypeError(f"cache_dir must be of type str or Path, not {type(cache_dir)}")
 
+    if not cache_conf:
+        raise ValueError(f"Missing cache")
+
+    if not isinstance(cache_conf, dict):
+        raise TypeError(
+            f"Invalid type for [cache_conf]: ({type(cache_conf)}). Must be of type dict."
+        )
+
     try:
-        _cache: diskcache.core.Cache = Cache(directory=cache_dir)
+        # _cache: diskcache.core.Cache = Cache(directory=cache_dir)
+        _cache: diskcache.core.Cache = Cache(**cache_conf)
 
         if index:
             cache_tag_index(cache=_cache)
@@ -91,15 +125,6 @@ def get_cache(
         raise Exception(f"Unhandled exception creating cache. Details: {exc}")
 
 
-## Define a default cache object
-default_cache: Cache = Cache(
-    directory=default_cache_dir,
-    timeout=convert_to_seconds(
-        unit=default_timeout_dict["unit"], amount=default_timeout_dict["amount"]
-    ),
-)
-
-
 def clear_cache(cache: Cache = None) -> bool:
     validate_cache(cache)
 

red_utils/httpx_utils/__init__.py

@@ -5,4 +5,4 @@
 from .validators import validate_client, validate_headers, validate_method
 from .operations import merge_headers, update_headers, get_req_client, make_request
 
-from validators import validate_client, validate_headers, validate_method
+from .validators import validate_client, validate_headers, validate_method

red_utils/loguru_utils/__init__.py

@@ -21,5 +21,4 @@
     default_trace_log_file_sink,
 )
 
-from .uvicorn_override import setup_uvicorn_logging
 from .validators import validate_compression_str, validate_level, validate_logger

red_utils/loguru_utils/uvicorn_override.py

@@ -1,4 +1,9 @@
 from __future__ import annotations
 
 from .constants import default_serialize_dir
-from .operations import msgpack_deserialize, msgpack_serialize
+from .operations import (
+    msgpack_deserialize,
+    msgpack_serialize,
+    msgpack_serialize_file,
+    msgpack_deserialize_file,
+)

red_utils/main.py

@@ -46,9 +46,43 @@ def ensure_path(dir: Union[str, Path] = None) -> bool:
         return True
 
 
-def msgpack_serialize(
+def msgpack_serialize(_json: dict = None) -> dict[str, Union[bool, str, bytes, None]]:
+    """Serialize a Python dict to a msgpack string.
+
+    Return object will be a dict with 2 keys, 'success' and 'detail.'
+    Success is a bool indicator of serialize success.
+    Detail contains the 'message' key with the bytestring, as well as other
+        optional details to be returned.
+    """
+    if not _json:
+        raise ValueError("Missing Python dict data to serialize")
+
+    try:
+        packed = msgpack.packb(_json)
+
+        print(f"Packed message type ({type(packed)})")
+
+        return_obj = {"success": True, "detail": {"message": packed}}
+
+    except Exception as exc:
+        return_obj = {"success": False, "detail": {"message": f"{exc}"}}
+
+    return return_obj
+
+
+def msgpack_serialize_file(
     _json: dict = None, output_dir: str = default_serialize_dir, filename: str = None
 ) -> dict[str, Union[bool, str, dict[str, Union[str, dict]]]]:
+    """Serialize a Python dict to a msgpack file.
+
+    Accepts a dict input, with an optional output_dir and filename. Serialize the
+    msgpack data to file at output_dir/filename.
+
+    Return object will be a dict with 2 keys, 'success' and 'detail.'
+    Success is a bool indicator of serialize success.
+    Detail contains the 'message' key with the bytestring, as well as other
+        optional details to be returned.
+    """
     if not _json:
         raise ValueError("Missing Python dict data to serialize")
 
@@ -59,6 +93,8 @@ def msgpack_serialize(
 
     if filename.endswith(".msgpack"):
         filename.replace(".msgpack", "")
+    else:
+        filename = f"{filename}.msgpack"
 
     dir_exist = ensure_path(output_dir)
 
@@ -76,18 +112,21 @@ def msgpack_serialize(
             }
 
         except Exception as exc:
-            # log.error(
-            #     {"exception": "Unhandled exception writing msgpack."}, exc_info=True
-            # )
-
             return_obj = {"success": False, "detail": {"message": f"{exc}"}}
 
     return return_obj
 
 
-def msgpack_deserialize(
+def msgpack_deserialize_file(
     filename: str = None,
 ) -> dict[str, Union[bool, str, dict[str, Union[str, dict]]]]:
+    """Load serialized msgpack string from a file and return.
+
+    Return object will be a dict with 2 keys, 'success' and 'detail.'
+    Success is a bool indicator of serialize success.
+    Detail contains the 'message' key with the bytestring, as well as other
+        optional details to be returned.
+    """
     if not filename:
         raise ValueError("Must pass a file name/path to deserialize")
 
@@ -113,3 +152,35 @@ def msgpack_deserialize(
         return_obj = {"success": False, "detail": {"message": f"{exc}"}}
 
     return return_obj
+
+
+def msgpack_deserialize(
+    packed_str: bytes = None,
+) -> dict[str, Union[bool, str, dict[str, Union[str, dict]]]]:
+    """Load serialized msgpack string.
+
+    Returns a dict with"""
+    if not packed_str:
+        raise ValueError("Must pass a bytestring to deserialize")
+
+    if not isinstance(packed_str, bytes):
+        raise TypeError(
+            f"Invalid type for [packed_str]: ({type(packed_str)}). Must be of type bytestring"
+        )
+
+    try:
+        unpacked = msgpack.unpackb(packed_str)
+
+        return_obj = {
+            "success": True,
+            "detail": {
+                "message": unpacked,
+            },
+        }
+
+    except Exception as exc:
+        # log.error({"exception": "Unhandled exception reading msgpack."}, exc_info=True)
+
+        return_obj = {"success": False, "detail": {"message": f"{exc}"}}
+
+    return return_obj