"""
This module provides utilities to initialize metric fields, collect errors,
safely resolve filesystem paths (no symlinks), and a size-limited writer (LimitedWriter)
"""
import os
from typing import List, Any
from io import BytesIO
def init_metric_fields(fields: List[str]) -> dict[str, Any]:
"""
Initializes a dictionary with the provided field names, setting each field's value to "NA".
Additionally, it adds an "errors" key with an empty list as its value.
"""
result = {field: "NA" for field in fields}
result["errors"] = []
return result
def add_error(result: dict[str, Any],
module: str,
code: int,
message: str) -> None:
"""
Adds an error entry to the errors list in the provided result dictionary.
"""
result["errors"].append({
"module": module,
"error_code": code,
"error_message": message
})
def check_path(path: str):
"""
Checks and resolves the given file path, ensuring it exists and is not a symlink.
"""
if not path:
return path, ValueError("The provided path is empty.")
origin = path
while not os.path.lexists(path):
path = os.path.dirname(path)
if path == "." or not path:
return "", FileNotFoundError("Path not found: %s", origin)
try:
abs_path = os.path.abspath(path)
resolved_path = os.path.realpath(abs_path)
if abs_path != resolved_path:
return "", RuntimeError("Can't support symlinks")
except Exception as e:
error_msg = str(e)
return "", RuntimeError("Failed to get absolute path or real path: %s", error_msg)
try:
abs_origin = os.path.abspath(origin)
except Exception as e:
return "", RuntimeError(f"get the absolute path failed: {e}")
return abs_origin, None
class LimitedWriter:
"""
LimitedWriter is an in-memory writer with a fixed byte limit.
"""
def __init__(self, limit: int):
self._buf = BytesIO()
self._limit = limit
self._written = 0
def write(self, data: bytes) -> int:
"""
Writes data to the buffer, respecting the buffer's size limit.
"""
if self._written >= self._limit:
return 0
room = self._limit - self._written
chunk = data[:room]
n = self._buf.write(chunk)
self._written += n
return n
def get_buffer_bytes(self) -> bytes:
"""
Retrieves the current contents of the buffer as bytes.
"""
return self._buf.getvalue()