pocketutils.misc.smartio

Complexity

Total Complexity

23

Size/Duplication

Total Lines	140
Duplicated Lines	0 %

Importance

Changes

0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	CompressionSet.guess()	0	7	3
A	SmartIo.read_bytes()	0	7	1
A	CompressionSet.__or__()	0	2	1
A	CompressionSet.__getitem__()	0	11	2
A	CompressionSet.empty()	0	3	1
A	CompressionSet.__add__()	0	6	4
A	CompressionSet.__sub__()	0	3	1
A	SmartIo.read_text()	0	6	1
B	SmartIo.write()	0	18	7
A	SmartIo.tmp_path()	0	6	1

"""
Compression-aware reading and writing of files.
"""
from __future__ import annotations

import abc
import bz2
import gzip
import lzma
import os
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path, PurePath
from typing import Any, Self

The name Self does not seem to exist in module typing.

import brotli

Unable to import 'brotli'

import lz4.frame

Unable to import 'lz4.frame'

import snappy

Unable to import 'snappy'

import zstandard

Unable to import 'zstandard'

from pocketutils import WritePermissionsError

PathLike = str | PurePath


@dataclass(frozen=True)

Missing class docstring

class Compression(metaclass=abc.ABCMeta):
    name: str
    suffixes: set[str]
    compress: Callable[[bytes], bytes]
    decompress: Callable[[bytes], bytes]


def identity(x):

Argument name "x" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

Missing function or method docstring

    return x


@dataclass(frozen=True)

Missing class docstring

class CompressionSet:
    mapping: dict[str, Compression]

Value 'dict' is unsubscriptable

    @classmethod
    def empty(cls) -> Self:

Missing function or method docstring

        return CompressionSet({"": Compression("", set(), identity, identity)})

    def __add__(self, fmt: Compression):
        new = {fmt.name: fmt} | {s: fmt for s in fmt.suffixes}
        already = {v for k, v in self.mapping.items() if k in new}
        if len(already) > 1 or len(already) == 1 and already != {fmt}:
            raise ValueError(f"Keys from {fmt} already mapped to {already}")
        return CompressionSet(self.mapping | new)

    def __sub__(self, fmt: Compression):
        return CompressionSet(
            {k: v for k, v in self.mapping.items() if k != fmt.name and k not in fmt.suffixes}
        )

    def __or__(self, fmt: CompressionSet):
        return CompressionSet(self.mapping | fmt.mapping)

    def __getitem__(self, t: Compression | str) -> Compression:

Argument name "t" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        """
        Returns a FileFormat from a name (e.g. "gz" or "gzip").
        Case-insensitive.

        Example:
            ``Compression.of("gzip").suffix  # ".gz"``
        """
        if isinstance(t, Compression):
            return t
        return self.mapping[t]

    def guess(self, path: PathLike) -> Compression:

Missing function or method docstring

        if "." not in path.name:
            return self[""]
        try:
            return self[path.suffix]
        except KeyError:
            return self[""]


COMPRESSIONS = (
    CompressionSet.empty()
    + Compression("gzip", {".gz", ".gzip"}, gzip.compress, gzip.decompress)
    + Compression("brotli", {".bro", ".brotli"}, brotli.compress, brotli.decompress)
    + Compression("zstandard", {".zst", ".zstd"}, zstandard.compress, zstandard.decompress)
    + Compression("lz4", {".lz4"}, lz4.frame.compress, lz4.frame.decompress)
    + Compression("snappy", {".snappy"}, snappy.compress, snappy.decompress)
    + Compression("bzip2", {".bz2", ".bzip2"}, bz2.compress, bz2.decompress)
    + Compression("xz", {".xz"}, lzma.compress, lzma.decompress)
)


@dataclass(slots=True, frozen=True)

Missing class docstring

The keyword slots does not seem to exist for the function call.

class SmartIo:
    @classmethod
    def write(

Missing function or method docstring

        cls, data: Any, path: PathLike, *, atomic: bool = False, mkdirs: bool = False

Wrong hanging indentation before block (add 4 spaces).

    ) -> None:
        path = Path(path)
        compressed = COMPRESSIONS.guess(path).compress(data)
        if path.exists() and not path.is_file():
            raise WritePermissionsError(f"Path {path} is not a file", path=path)
        if path.exists() and not os.access(path, os.W_OK):
            raise WritePermissionsError(f"Cannot write to {path}", path=path)
        if mkdirs:
            path.parent.mkdir(parents=True, exist_ok=True)
        if atomic:
            tmp = cls.tmp_path(path)
            path.write_bytes(compressed)
            tmp.rename(path)
        else:
            path.write_bytes(compressed)

    @classmethod
    def read_text(cls, path: PathLike) -> str:
        """
        Similar to :meth:`read_bytes`, but then converts to UTF-8.
        """
        return cls.read_bytes(path).decode(encoding="utf-8")

    @classmethod
    def read_bytes(cls, path: PathLike) -> bytes:
        """
        Reads, decompressing according to the filename suffix.
        """
        data = Path(path).read_bytes()
        return COMPRESSIONS.guess(path).decompress(data)

    @classmethod
    def tmp_path(cls, path: PathLike, extra: str = "tmp") -> Path:

Missing function or method docstring

        now = datetime.now().isoformat(timespec="microsecond").replace(":", "").replace("-", "")
        path = Path(path)
        suffix = "".join(path.suffixes)
        return path.parent / f".part_{extra}.{now}{suffix}"


__all__ = ["Compression", "CompressionSet", "COMPRESSIONS", "SmartIo"]