File I/O

utils

File I/O utilities for CSV/JSON serialization, path generation, and OS file opening.

flatten_dict

flatten_dict(*, d: Dict[str, Any], parent_key: str = '', sep: str = '.') -> Dict[str, Any]

Recursively flattens a nested dictionary using dot notation.

Parameters:

Name	Type	Description	Default
d	Dict[str, Any]	The dictionary to flatten.	required
parent_key	str	Prefix for keys at the current nesting level (used during recursion).	''
sep	str	Separator between parent and child key segments.	'.'

Returns:

Type	Description
Dict[str, Any]	A single-level dictionary with composite keys joined by sep.

Example

{"a": {"b": 1}} -> {"a.b": 1}

Source code in t3api_utils/file/utils.py

def flatten_dict(
    *, d: Dict[str, Any], parent_key: str = "", sep: str = "."
) -> Dict[str, Any]:
    """Recursively flattens a nested dictionary using dot notation.

    Args:
        d: The dictionary to flatten.
        parent_key: Prefix for keys at the current nesting level (used during
            recursion).
        sep: Separator between parent and child key segments.

    Returns:
        A single-level dictionary with composite keys joined by ``sep``.

    Example:
        {"a": {"b": 1}} -> {"a.b": 1}
    """
    result: Dict[str, Any] = {}
    for k, v in d.items():
        new_key = f"{parent_key}{sep}{k}" if parent_key else k
        if isinstance(v, dict):
            result.update(flatten_dict(d=v, parent_key=new_key, sep=sep))
        else:
            result[new_key] = v
    return result

prioritized_fieldnames

prioritized_fieldnames(*, dicts: List[Dict[str, Any]]) -> List[str]

Orders CSV fieldnames by priority list, then appends remaining keys alphabetically.

Parameters:

Name	Type	Description	Default
dicts	List[Dict[str, Any]]	List of dictionaries whose keys represent fieldnames.	required

Returns:

Type	Description
List[str]	An ordered list of fieldnames with priority fields first, followed by
List[str]	any remaining keys in alphabetical order.

Source code in t3api_utils/file/utils.py

def prioritized_fieldnames(*, dicts: List[Dict[str, Any]]) -> List[str]:
    """Orders CSV fieldnames by priority list, then appends remaining keys alphabetically.

    Args:
        dicts: List of dictionaries whose keys represent fieldnames.

    Returns:
        An ordered list of fieldnames with priority fields first, followed by
        any remaining keys in alphabetical order.
    """
    all_keys = {key for row in dicts for key in row.keys()}
    prioritized = [key for key in PRIORITY_FIELDS if key in all_keys]
    remaining = sorted(all_keys - set(prioritized))
    return prioritized + remaining

default_json_serializer

default_json_serializer(*, obj: object) -> str

Fallback serializer for non-JSON-native types.

Parameters:

Name	Type	Description	Default
obj	object	The object that json.dump could not serialize natively.	required

Returns:

Type	Description
str	An ISO-format string if obj is a datetime instance.

Raises:

Type	Description
TypeError	If the object type is not supported for serialization.

Source code in t3api_utils/file/utils.py

def default_json_serializer(*, obj: object) -> str:
    """Fallback serializer for non-JSON-native types.

    Args:
        obj: The object that ``json.dump`` could not serialize natively.

    Returns:
        An ISO-format string if ``obj`` is a ``datetime`` instance.

    Raises:
        TypeError: If the object type is not supported for serialization.
    """
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")

generate_output_path

generate_output_path(*, model_name: str, license_number: str, output_dir: str, extension: str) -> Path

Constructs the output file path with consistent naming.

The filename format is <model_name>__<license_number>__<timestamp>.<ext>. Parent directories are created automatically if they do not exist.

Parameters:

Name	Type	Description	Default
model_name	str	Logical model name included in the filename.	required
license_number	str	License identifier included in the filename.	required
output_dir	str	Directory where the file will be saved.	required
extension	str	File extension without the leading dot (e.g., "csv").	required

Returns:

Type	Description
Path	A Path to the generated output file.

Source code in t3api_utils/file/utils.py

def generate_output_path(
    *, model_name: str, license_number: str, output_dir: str, extension: str
) -> Path:
    """Constructs the output file path with consistent naming.

    The filename format is ``<model_name>__<license_number>__<timestamp>.<ext>``.
    Parent directories are created automatically if they do not exist.

    Args:
        model_name: Logical model name included in the filename.
        license_number: License identifier included in the filename.
        output_dir: Directory where the file will be saved.
        extension: File extension without the leading dot (e.g., ``"csv"``).

    Returns:
        A ``Path`` to the generated output file.
    """
    timestamp = datetime.now().isoformat(timespec="seconds").replace(":", "-")
    filename = f"{model_name}__{license_number}__{timestamp}.{extension}"
    filepath = Path(output_dir) / filename
    filepath.parent.mkdir(parents=True, exist_ok=True)
    return filepath

save_dicts_to_json

save_dicts_to_json(*, dicts: List[Dict[str, Any]], model_name: str, license_number: str, output_dir: str = 'output') -> Path

Saves a list of dictionaries to a JSON file.

Parameters:

Name	Type	Description	Default
dicts	List[Dict[str, Any]]	Non-empty list of dictionaries to serialize.	required
model_name	str	Logical model name used in the output filename.	required
license_number	str	License identifier used in the output filename.	required
output_dir	str	Directory where the JSON file will be saved.	'output'

Returns:

Type	Description
Path	The Path to the written JSON file.

Raises:

Type	Description
ValueError	If dicts is empty.

Source code in t3api_utils/file/utils.py

def save_dicts_to_json(
    *,
    dicts: List[Dict[str, Any]],
    model_name: str,
    license_number: str,
    output_dir: str = "output",
) -> Path:
    """Saves a list of dictionaries to a JSON file.

    Args:
        dicts: Non-empty list of dictionaries to serialize.
        model_name: Logical model name used in the output filename.
        license_number: License identifier used in the output filename.
        output_dir: Directory where the JSON file will be saved.

    Returns:
        The ``Path`` to the written JSON file.

    Raises:
        ValueError: If ``dicts`` is empty.
    """
    if not dicts:
        raise ValueError("Input list is empty")

    filepath = generate_output_path(model_name=model_name, license_number=license_number, output_dir=output_dir, extension="json")

    with open(filepath, "w", encoding="utf-8") as f:
        json.dump(
            dicts, f, ensure_ascii=False, indent=2, default=lambda obj: default_json_serializer(obj=obj)
        )

    logger.info(f"Wrote {len(dicts)} {model_name} objects to {filepath}")
    return filepath

save_dicts_to_csv

save_dicts_to_csv(*, dicts: List[Dict[str, Any]], model_name: str, license_number: str, output_dir: str = 'output', strip_empty_columns: bool = False) -> Path

Saves a list of (possibly nested) dictionaries to a CSV file after flattening.

Parameters:

Name	Type	Description	Default
dicts	List[Dict[str, Any]]	Non-empty list of input dictionaries (may contain nested dicts).	required
model_name	str	Logical model name used in the output filename.	required
license_number	str	License identifier used in the output filename.	required
output_dir	str	Directory where the CSV file will be saved.	'output'
strip_empty_columns	bool	If True, columns where every value is None, "", or [] will be removed.	False

Returns:

Type	Description
Path	The Path to the written CSV file.

Raises:

Type	Description
ValueError	If dicts is empty.

Source code in t3api_utils/file/utils.py

def save_dicts_to_csv(
    *,
    dicts: List[Dict[str, Any]],
    model_name: str,
    license_number: str,
    output_dir: str = "output",
    strip_empty_columns: bool = False,
) -> Path:
    """Saves a list of (possibly nested) dictionaries to a CSV file after flattening.

    Args:
        dicts: Non-empty list of input dictionaries (may contain nested dicts).
        model_name: Logical model name used in the output filename.
        license_number: License identifier used in the output filename.
        output_dir: Directory where the CSV file will be saved.
        strip_empty_columns: If True, columns where every value is ``None``,
            ``""``, or ``[]`` will be removed.

    Returns:
        The ``Path`` to the written CSV file.

    Raises:
        ValueError: If ``dicts`` is empty.
    """
    if not dicts:
        raise ValueError("Input list is empty")

    flat_dicts = [flatten_dict(d=d) for d in dicts]

    if strip_empty_columns:
        # Determine which fields are completely empty
        all_keys = {key for d in flat_dicts for key in d}
        non_empty_keys = {
            key
            for key in all_keys
            if any(d.get(key) not in (None, "", []) for d in flat_dicts)
        }
        flat_dicts = [
            {k: v for k, v in d.items() if k in non_empty_keys} for d in flat_dicts
        ]

    fieldnames = prioritized_fieldnames(dicts=flat_dicts)
    filepath = generate_output_path(model_name=model_name, license_number=license_number, output_dir=output_dir, extension="csv")

    with open(filepath, mode="w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(flat_dicts)

    logger.info(f"Wrote {len(flat_dicts)} {model_name} objects to {filepath}")
    return filepath

open_file

open_file(*, path: Path) -> None

Opens a file using the default application for the current OS.

Parameters:

Name	Type	Description	Default
path	Path	Filesystem path to the file to open.	required

Source code in t3api_utils/file/utils.py

def open_file(*, path: Path) -> None:
    """Opens a file using the default application for the current OS.

    Args:
        path: Filesystem path to the file to open.
    """
    try:
        if sys.platform == "darwin":
            subprocess.run(["open", str(path)], check=False)
        elif os.name == "nt":
            os.startfile(path)
        elif os.name == "posix":
            subprocess.run(["xdg-open", str(path)], check=False)
    except Exception as e:
        logger.error(f"Failed to open file: {e}")