4
\$\begingroup\$

The first edition of this code was written back when Python 2 had not yet been deprecated in favor of Python 3. The difference between strings and bytes were not very clear back then. Also, PEP8 violations abound in the original version.

'''Module for Directory and File Serialization.

This module provides two classes that implement the
DFS (Directory and File Serialization) file format.'''

__version__ = '1.0'

import os as _os
import sys as _sys

################################################################################

FORMAT_DOCUMENTATION = '''\
Directory
    Header
        0,aaa,b,c,dd
            0 = Directory
            a = Pointer Size (bytes)
            b = Name Size (bytes)
            c = Content Flag
            d = Type Code
                00 = End
                01 = Error
                10 = Error
                11 = Real
    Pointer
    Name Size
    Name

File
    Header
        1,aaa,b,ccc
            1 = File
            a = Pointer Size (bytes)
            b = Name Size (bytes)
            c = Data Size (bytes)
    Pointer
    Name Size
    Name
    Data Size
    Data'''

################################################################################

class Acquire:

    'Acquire(destination) -> Acquire'

    BUFF_SIZE = 2 ** 20

    def __init__(self, destination):
        'Initialize the Acquire object.'
        self.__destination = destination
        self.__destination_path = _os.path.abspath(destination.name) if hasattr(destination, 'name') else None
        self.__archive = False

    def acquire(self, source):
        'Save source to destination.'
        source = _os.path.abspath(source)
        self.__pointer = 0
        if self.__archive:
            self.__destination.write('\0')
        else:
            self.__archive = True
        if _os.path.isdir(source):
            self.__dir(source, '\0')
        elif _os.path.isfile(source):
            if source == self.__destination_path:
                raise ValueError, 'Source cannot be destination.'
            self.__file(source, '\0')
        else:
            raise ValueError, 'Source must be directory or file.'

    def __dir(self, source, pointer):
        'Private class method.'
        name = _os.path.basename(source)
        name_size = _str(len(name))
        try:
            dir_list = _os.listdir(source)
        except:
            dir_list = ()
        if dir_list:
            self.__pointer += 1
            self.__destination.write(chr((len(pointer) - 1 << 4) + (len(name_size) - 1 << 3) + 7) + pointer + name_size + name)
        else:
            self.__destination.write(chr((len(pointer) - 1 << 4) + (len(name_size) - 1 << 3) + 3) + pointer + name_size + name)
        pointer = _str(self.__pointer)
        for name in dir_list:
            source_name = _os.path.join(source, name)
            if source_name == self.__destination_path:
                continue
            elif _os.path.isdir(source_name):
                self.__dir(source_name, pointer)
            elif _os.path.isfile(source_name):
                self.__file(source_name, pointer)

    def __file(self, source, pointer):
        'Private class method.'
        name = _os.path.basename(source)
        name_size = _str(len(name))
        try:
            source = open(source, 'rb')
            source.seek(0, 2)
            data_size = _str(source.tell())
            source.seek(0, 0)
        except:
            source = None
            data_size = '\0'
        self.__destination.write(chr(127 + (len(pointer) - 1 << 4) + (len(name_size) - 1 << 3) + len(data_size)) + pointer + name_size + name + data_size)
        if source:
            data_size = _int(data_size)
            try:
                while data_size:
                    buff = source.read(min(self.BUFF_SIZE, data_size))
                    if buff:
                        self.__destination.write(buff)
                        data_size -= len(buff)
                    else:
                        raise IOError, 'File changed size while open.'
                position = source.tell()
                source.seek(0, 2)
                if position != source.tell():
                    raise IOError, 'File changed size while open.'
            finally:
                source.close()

################################################################################

class Release:

    'Release(source) -> Release'

    BUFF_SIZE = 2 ** 20

    def __init__(self, source):
        'Initialize the Release object.'
        self.__source = source
        self.__EOF = False

    def release(self, destination):
        'Save source to destination.'
        if self.__EOF:
            raise EOFError, 'End Of File Found'
        self.__parents = [_os.path.abspath(destination)]
        header = self.__source.read(1)
        header = ord(header) if header else -1
        if header == -1:
            self.__EOF = True
            raise Warning, 'Irregular File Termination Detected'
        while header != -1 and (header > 127 or header & 3):
            if header < 128:
                if header & 3 != 3:
                    raise IOError, 'Corrupt Directory Header Found'
                self.__dir(header)
            else:
                self.__file(header)
            header = self.__source.read(1)
            header = ord(header) if header else -1
        if header == -1:
            self.__EOF = True

    def EOF(self):
        'Return the End Of File status.'
        return self.__EOF

    def __dir(self, header):
        'Private class method.'
        path = _os.path.join(self.__parents[_int(self.__read((header >> 4 & 7) + 1))], self.__read(_int(self.__read((header >> 3 & 1) + 1))))
        _os.mkdir(path)
        if header >> 2 & 1:
            self.__parents.append(path)

    def __file(self, header):
        'Private class method.'
        destination = open(_os.path.join(self.__parents[_int(self.__read((header >> 4 & 7) + 1))], self.__read(_int(self.__read((header >> 3 & 1) + 1)))), 'wb')
        data_size = _int(self.__read((header & 7) + 1))
        try:
            while data_size:
                buff = self.__source.read(min(self.BUFF_SIZE, data_size))
                if buff:
                    destination.write(buff)
                    data_size -= len(buff)
                else:
                    raise IOError, 'End Of File Found'
        finally:
            destination.close()

    def __read(self, size):
        'Private class method.'
        if size:
            buff = ''
            while size:
                temp = self.__source.read(size)
                if temp:
                    buff += temp
                    size -= len(temp)
                else:
                    raise IOError, 'End Of File Found'
            return buff
        raise IOError, 'Zero Length String Found'

################################################################################

def _str(integer):
    'Private module function.'
    if integer:
        string = ''
        while integer:
            string = chr(integer & 0xFF) + string
            integer >>= 8
        return string
    return '\0'

def _int(string):
    'Private module function.'
    integer = 0
    for c in string:
        integer <<= 8
        integer += ord(c)
    return integer

################################################################################

if __name__ == '__main__':
    _sys.stdout.write('Content-Type: text/plain\n\n')
    _sys.stdout.write(file(_sys.argv[0]).read())

The code has since then been rewritten for Python 3 and attempts to take advantage of its many changes. There is clarity between what should be bytes and what should be strings, and no PEP8 problems should be found in the new version.

#! /usr/bin/env python3
"""Provide a simple directory and file serialization protocol.

This module implements two classes that can handle the DFS (Directory &
File Serialization) file format. Both classes can deal with file-like
objects and stream directories and files to and from the file system."""


# Import other modules needed for this module to work properly.
import abc
import collections
import enum
import io
import pathlib


# Include supplemental information along with a public API definition.
__author__ = 'Stephen "Zero" Chappell <Noctis.Skytower@gmail.com>'
__date__ = '9 February 2017'
__version__ = 3, 0, 0
__all__ = 'Serializer', 'Deserializer'


# The organization of the serialized data is fairly simple as shown below.
SERIALIZATION_FORMAT = '''\
Directory
    Header
        0,aaa,b,c,dd (Bit Mapping)
            0 = Directory
            a = Pointer Length
            b = Name Size Length
            c = Content Flag
            d = Type Code
                00 = Separator
                01 = Reserved
                10 = Reserved
                11 = Genuine
    Pointer to Parent
    Name Size
    Name
---------------------------------
File
    Header
        1,aaa,b,ccc (Bit Mapping)
            1 = File
            a = Pointer Length
            b = Name Size Length
            c = Data Size Length
    Pointer to Parent
    Name Size
    Name
    Data Size
    Data
'''


@enum.unique
class _RecordType(enum.IntEnum):

    """Enumeration of the different types a record may represent."""

    DIRECTORY = 0b0
    FILE = 0b1


@enum.unique
class _DirectoryTypeCode(enum.IntEnum):

    """Enumeration of codes directories may specify for their type."""

    SEPARATOR = 0b00
    RESERVED_A = 0b01
    RESERVED_B = 0b10
    GENUINE = 0b11


# Define the necessary components used to describe a bit field.
_BitField = collections.namedtuple('_BitField', 'offset, width')


class _Common(abc.ABC):

    """Abstract class for supporting Serializer and Deserializer classes."""

    # Define a few static attributes for use in derived classes.
    BUFFER_SIZE = 1 << 20
    BYTE_WIDTH = 8
    BYTE_MASK = (1 << BYTE_WIDTH) - 1
    NAME_ENCODING = 'utf_8'     # Set to 'mbcs' for Archive 2.0 compatibility.
    NULL_BYTE = b'\0'

    # Define the bit fields used in header bytes.
    RECORD_TYPE = _BitField(7, 1)
    POINTER_LENGTH = _BitField(4, 3)
    NAME_SIZE_LENGTH = _BitField(3, 1)
    CONTENT_FLAG = _BitField(2, 1)
    DIRECTORY_TYPE_CODE = _BitField(0, 2)
    FILE_DATA_SIZE_LENGTH = _BitField(0, 3)

    @abc.abstractmethod
    def __init__(self, stream):
        """Initialize the _Common instance's attributes."""
        self._stream = stream
        self._header = None

    @classmethod
    def _int_to_bytes(cls, integer):
        """Convert a number into a byte string of variable length."""
        if integer:
            array = bytearray()
            while integer:
                array.insert(0, integer & cls.BYTE_MASK)
                integer >>= cls.BYTE_WIDTH
            return bytes(array)
        return cls.NULL_BYTE

    @classmethod
    def _bytes_to_int(cls, array):
        """Convert a byte string of variable length into a number."""
        integer = 0
        for byte in array:
            integer <<= cls.BYTE_WIDTH
            integer |= byte
        return integer

    @staticmethod
    def _write(file, buffer):
        """Write buffer to file until it is completely written."""
        while True:
            written = file.write(buffer)
            if written is None:
                raise IOError('nothing could be written to the file')
            if written == len(buffer):
                break
            buffer = buffer[written:]


class Serializer(_Common):

    """Serializer(destination) -> Serializer instance"""

    def __init__(self, destination):
        """Initialize the Serializer instance's attributes."""
        super().__init__(destination)
        self._started = False
        self._pointer = None

    def run(self, source, keep_zombies=True):
        """Dump the source file or directory contents onto the destination."""
        path = pathlib.Path(source).resolve()
        zombies = []
        if path.is_dir():
            self._prime_run()
            self._acquire_dir(path, self.NULL_BYTE, keep_zombies, zombies)
        elif path.is_file():
            self._prime_run()
            self._acquire_file(path, self.NULL_BYTE, keep_zombies, zombies)
        else:
            raise ValueError('source must be a dir or a file')
        return zombies

    def _prime_run(self):
        """Reset some attributes before a serialization run."""
        self._pointer = 0
        if self._started:
            self._write(self._stream, self.NULL_BYTE)
        else:
            self._started = True

    def _acquire_dir(self, source, parent, keep_zombies, zombies):
        """Serialize a directory."""
        try:
            paths = tuple(source.iterdir())
        except OSError:
            zombies.append(source)
            if not keep_zombies:
                return
            paths = ()
        self._write_complete_dir_header(source, parent, bool(paths))
        if paths:
            self._pointer += 1
            parent = self._int_to_bytes(self._pointer)
            for path in paths:
                if path.is_dir():
                    self._acquire_dir(path, parent, keep_zombies, zombies)
                elif path.is_file():
                    self._acquire_file(path, parent, keep_zombies, zombies)

    def _write_complete_dir_header(self, source, parent, content):
        """Record all directory information except its contents."""
        name = source.name.encode(self.NAME_ENCODING)
        name_size = self._int_to_bytes(len(name))
        self._write_dir_header_byte(parent, name_size, content)
        self._write(self._stream, parent)
        self._write(self._stream, name_size)
        self._write(self._stream, name)

    def _write_dir_header_byte(self, pointer, name_size, content):
        """Record the directory header byte using the correct format."""
        self._header = 0
        self._set_bits(_RecordType.DIRECTORY, self.RECORD_TYPE)
        self._set_bits(len(pointer) - 1, self.POINTER_LENGTH)
        self._set_bits(len(name_size) - 1, self.NAME_SIZE_LENGTH)
        self._set_bits(content, self.CONTENT_FLAG)
        self._set_bits(_DirectoryTypeCode.GENUINE, self.DIRECTORY_TYPE_CODE)
        self._write(self._stream, bytes([self._header]))

    def _set_bits(self, integer, bit_field):
        """Help build the header byte while checking certain arguments."""
        if not 0 <= integer < 1 << bit_field.width:
            raise ValueError('integer does not fit in width numbers of bits')
        self._header |= integer << bit_field.offset

    def _acquire_file(self, source, parent, keep_zombies, zombies):
        """Serialize a file."""
        restore_point = self._stream.tell()
        try:
            with source.open('rb') as file:
                file_length = file.seek(0, io.SEEK_END)
                self._write_complete_file_header(source, parent, file_length)
                future_data = file.seek(0, io.SEEK_END)
                if future_data != file_length:
                    raise OSError('source changed size after writing header')
                file.seek(0, io.SEEK_SET)
                while future_data:
                    buffer = file.read(min(future_data, self.BUFFER_SIZE))
                    if not buffer:
                        raise OSError('source file ended with remaining data')
                    self._write(self._stream, buffer)
                    future_data -= len(buffer)
                if file.seek(0, io.SEEK_END) != file_length:
                    raise OSError('file changed size during serialization')
        except OSError:
            self._stream.seek(restore_point, io.SEEK_SET)
            self._stream.truncate()
            zombies.append(source)
            if keep_zombies:
                self._write_complete_file_header(source, parent, 0)

    def _write_complete_file_header(self, source, parent, file_length):
        """Record all file information except its data."""
        name = source.name.encode(self.NAME_ENCODING)
        name_size = self._int_to_bytes(len(name))
        data_size = self._int_to_bytes(file_length)
        self._write_file_header_byte(parent, name_size, data_size)
        self._write(self._stream, parent)
        self._write(self._stream, name_size)
        self._write(self._stream, name)
        self._write(self._stream, data_size)

    def _write_file_header_byte(self, pointer, name_size, data_size):
        """Record the file header byte using the correct format."""
        self._header = 0
        self._set_bits(_RecordType.FILE, self.RECORD_TYPE)
        self._set_bits(len(pointer) - 1, self.POINTER_LENGTH)
        self._set_bits(len(name_size) - 1, self.NAME_SIZE_LENGTH)
        self._set_bits(len(data_size) - 1, self.FILE_DATA_SIZE_LENGTH)
        self._write(self._stream, bytes([self._header]))


class Deserializer(_Common):

    """Deserializer(source) -> Deserializer instance"""

    def __init__(self, source):
        """Initialize the Deserializer instance's attributes."""
        super().__init__(source)
        self._finished = False
        self._parents = None

    @property
    def finished(self):
        """Check if the object has reached the end of the file yet."""
        return self._finished

    def run(self, destination):
        """Load the source file-like object onto the destination directory."""
        if self._finished:
            raise EOFError('end of file was found')
        self._parents = [pathlib.Path(destination).resolve()]
        starting_run = True
        while True:
            byte = self._stream.read(1)
            if not byte:
                self._finished = True
                if starting_run:
                    raise IOError('unexpected file termination detected')
                break
            self._header = byte[0]
            if self._get_bits(self.RECORD_TYPE) == _RecordType.FILE:
                self._release_file()
            else:
                type_code = self._get_bits(self.DIRECTORY_TYPE_CODE)
                if type_code == _DirectoryTypeCode.GENUINE:
                    self._release_dir()
                elif type_code == _DirectoryTypeCode.SEPARATOR:
                    if starting_run:
                        raise IOError('empty record detected')
                    break
                else:
                    raise IOError('reserved directory type code detected')
            starting_run = False

    def _get_bits(self, bit_field):
        """Extract width number of bits from header starting at offset."""
        return self._header >> bit_field.offset & (1 << bit_field.width) - 1

    def _release_dir(self):
        """Deserialize a directory."""
        pointer_length = self._get_bits(self.POINTER_LENGTH) + 1
        name_size_length = self._get_bits(self.NAME_SIZE_LENGTH) + 1
        content_flag = bool(self._get_bits(self.CONTENT_FLAG))
        # After decoding the header byte, read and process the remaining data.
        pointer = self._bytes_to_int(self._read(pointer_length))
        name_size = self._bytes_to_int(self._read(name_size_length))
        name = self._read(name_size).decode(self.NAME_ENCODING)
        path = self._parents[pointer] / name
        path.mkdir()
        if content_flag:
            self._parents.append(path)

    def _release_file(self):
        """Deserialize a file."""
        pointer_length = self._get_bits(self.POINTER_LENGTH) + 1
        name_size_length = self._get_bits(self.NAME_SIZE_LENGTH) + 1
        data_size_length = self._get_bits(self.FILE_DATA_SIZE_LENGTH) + 1
        # After decoding the header byte, read and process the remaining data.
        pointer = self._bytes_to_int(self._read(pointer_length))
        name_size = self._bytes_to_int(self._read(name_size_length))
        name = self._read(name_size).decode(self.NAME_ENCODING)
        with (self._parents[pointer] / name).open('wb') as destination:
            future_data = self._bytes_to_int(self._read(data_size_length))
            while future_data:
                buffer = self._stream.read(min(future_data, self.BUFFER_SIZE))
                if not buffer:
                    raise IOError('end of file was found')
                self._write(destination, buffer)
                future_data -= len(buffer)

    def _read(self, future_data):
        """Read at least as many bytes from the source as requested."""
        if future_data:
            buffer = bytearray()
            while future_data:
                data = self._stream.read(future_data)
                if not data:
                    raise IOError('end of file was found')
                buffer.extend(data)
                future_data -= len(data)
            return buffer
        raise IOError('request for zero bytes found')

Please note that EOFError should only be used when the end-of-file has been reached in an expected manner. Otherwise, most errors should be considered to be IOError in nature. Can you identify any areas to improve the second rewrite?

\$\endgroup\$
  • 2
    \$\begingroup\$ For the record, Python 2 is still not deprecated and is still supported widely heh. \$\endgroup\$ – Thomas Ward Feb 20 '17 at 16:25
  • \$\begingroup\$ Thank you for pointing this out! It should have been noted that the deprecation was personal and not official. After reading about the backwards incompatible changes in Python 3, I avoiding personally programming in Python for over a year until after it was released. \$\endgroup\$ – Noctis Skytower Feb 20 '17 at 16:39
  • \$\begingroup\$ Since you are asking about a review for the python 3 compliant code only, I would recommend removing the python 2 tag from the question. \$\endgroup\$ – dfhwze Aug 3 at 20:00
3
\$\begingroup\$

This is a review of your 2nd review request, described as a rewrite for python3.

Summary: code looks great! Much better than the older version. Many changes I suggested for the old already appear in the new. Thank you for turning mangling double __ into private single _ prefix.

        0,aaa,b,c,dd (Bit Mapping)

The explicit "bit mapping" remark is very helpful, thank you, a definite improvement in clarity over v1.

    """Enumeration of the different types a record may represent."""
    ...
    """Enumeration of codes directories may specify for their type."""
    ...
    """Initialize the _Common instance's attributes."""
    ...
    """Initialize the Serializer instance's attributes."""

These docstrings are obvious so they don't really help out. Consider deleting them. I am not keen to write comments or docstrings that are redundant with what identifiers tell us, for fear the code will change and the remarks will lag behind (DRY). The code should tell what is specific, and remarks should offer vague generalities that aid comprehension.

@enum.unique

Thank you, that is helpful.

    POINTER_LENGTH = _BitField(4, 3)

Thank you, this is a much better way of explaining the meaning of each bit. Very clear. The shifting and bit twiddling is similarly much improved, and it enjoys appropriate use of descriptive identifiers.

    def _write(file, buffer):
        ...
        while True:

This method works and is clear. There is an "early out" break for the normal case of whole thing written. That seems like a premature optimization. The natural looping condition would be what the docstring explains: loop until written. If we took out the break then the normal case would assign buffer = buffer[written:], which works out to buffer = []. Seems pretty cheap to execute. Then the looping condition would ask if bytes remain, notice it's empty, and terminate.

Separate item. I'm looking at https://docs.python.org/3/library/io.html#io.RawIOBase.write . It explains:

None is returned if the raw stream is set not to block and no single byte could be readily written to it.

The code correctly throws if that happens. But it seems hard to provoke that -- it's not clear to me your file will ever be set to non-blocking. I am reluctant to ship code that never executed, even if it is simple. Consider deleting that clause. (Put another way, I have found it very embarrassing to ship an untested syntax error in my error handler.)

    """Serializer(destination) -> Serializer instance"""

Sorry, I'm still not understanding that. PEP-8 asks for an English sentence here, and I agree with the standard's sentiment, that a full sentence may be more broadly understood than powerpoint bullet points. The only sense I can make of this would be appropriate for the ctor, which has a different (obvious) docstring, and in any event this docstring would be obvious and convey nothing new, as it simply says what a constructor does.

I'm looking at def run(). Now, maybe there's some parallel construction reason for using run in both Serializer and Deserializer that I'm not yet seeing. But, just reading your (very nice) docstring, dump() would be the natural verb to use here.

Using Path is very nice. The expression pathlib.Path(source) is slightly verbose -- it is usual to from pathlib import Path.

    def run(self, source, keep_zombies=True):
        """Dump the source file or directory contents onto the destination."""

This is the first docstring I found inadequate. Your audience is developers wanting to correctly

  1. call your method
  2. interpret its results

I honestly don't know what this whole zombies concept is about. (It makes me think of a forked child that called exit but parent has not yet wait4ed, and I'm not yet seeing how that would be relevant.) Please have the docstring, or at least a comment, offer some guidance about why I might choose a non-default keep_zombies setting, and what the return value is for.

        if self._started:
            self._write(self._stream, self.NULL_BYTE)

Sorry, I'm sure this relates to the SERIALIZATION_FORMAT spec above, but I'm not yet seeing how a NULL_BYTE delimits one section from another. That is, I find this code obscure, I don't see how it fits with the spec.

An alternate approach might have separator be an empty bytestring or one that holds a NULL_BYTE, so we can unconditionally write out separator.

        """Serialize a directory."""
        ...
        """Serialize a file."""

The docstrings are clear and helpful. They should be unnecessary, redundant with the method name, except that curiously the method starts "acquire" rather than "serialize". Not sure what the motivation was.

            paths = tuple(source.iterdir())

This is not pythonic. An arbitrary number of directory entries will come back. Prefer list over tuple in such a case. Assign [] at end of exception handler.

        except OSError:

I wouldn't mind a # comment about "permission error", if that is the chief concern here.

Ok, the concept of "zombies" is starting to become clearer. An alternate name might be "failed transfers", or maybe "denied files" / "unreadable files".

        """Record the directory header byte using the correct format."""

The last part seems redundant -- it's not like we would try to do it incorrectly.

We're writing many bytes, so prolly plural would be appropriate: def _write_dir_header_bytes

                file.seek(0, io.SEEK_SET)

I don't understand what this does. It appears to be a no-op.

There's a bunch of seeking here. It's not obvious to me why that's more convenient than repeatedly stating the file to see if things changed.

    def finished(self):

This is nice enough. Consider using a prefix with such a boolean predicate: is_finished()

documentation

Tell us about zombies and their use cases.

Tell us about racy behavior, such as when we serialize /var/log/messages and syslogd is appending to it.

tests

Supply unit tests that exercise the code.

algorithm

Consider relying on stat for current size, rather than seek. (Calling stat obviously doesn't change your current location.)

You have quite a few ifs that can raise. It seems "hard" to trigger some of those. The caller still has to expect that a Bad Thing can happen at any point, lightning might strike, the filesystem could explode. Rather than doing checks and raising your own error, consider letting some FS errors just naturally propagate up, and the caller can figure out which ones are important for him to catch.

Or, perhaps you could wrap such an error handler around your layer so several trys become just one. One is much easier to test!

\$\endgroup\$
3
\$\begingroup\$

DFS (Directory and File Serialization) file format

Not sure what that is. Nor is google. Recommend you offer a more specific citation, hopefully one that includes an URL.

import os as _os
import sys as _sys

Hmmm, that's odd. Do you really need to rename them?

FORMAT_DOCUMENTATION = '''\

The backwhack is very odd -- don't do it. That's what triple quotes are for.

        0 = Directory

You apparently meant "0 = type is Directory". Similarly for "1 = File".

        b = Name Size (bytes)

What?!? You're apparently indicating that b is a single bit, and it denotes this many vs that many bytes?!? How many bytes?

    'Acquire(destination) -> Acquire'

I understand that you think that's a docstring. I don't understand what it's trying to tell me. Same remark for Release.

    'Initialize the Acquire object.'

Putting such a docstring on a ctor is redundant; consider deleting it. Same remark for Release.

    self.__destination = destination

Are you sure you want name mangling? Consider using just a single leading _ underscore for these three. Same remark for Release.

    self.__pointer = 0

You should probably init this to zero in the ctor, as well. It gives the reader a hint about which balls will be up in the air during the lifetime of this object.

The meaning of self.__archive is not transparently obvious to me.

def __dir(self, source, pointer):

Oh, goodness! Stop with the name mangling already! Is there some inheritance use case coded up where mangling helps? Just use a single _ underscore.

    'Private class method.'

That's not a docstring, it's a # comment. It doesn't help the caller figure out how to correctly call this private method. It is clearly not a @classmethod, rather, each instantiated object owns it. Similarly, delete the 'Private module function.' docstrings; they're not helpful.

        self.__pointer += 1

There is a pre-condition here: acquire() must execute before __dir() does. That is not at all obvious. Make it so, by initializing pointer in the ctor.

        self.__destination.write(chr((len(pointer) - 1 << 4) + (len(name_size) - 1 << 3) + 3) + pointer + name_size + name)

A single digit, 3, is different from the preceding line, where it was 7. Rather than pasting the line in twice, parameterize it, highlighting the 3 / 7 contrast.

I was rather hoping to see a call to pack(), but perhaps the bit packing is not a good fit for that.

        source.seek(0, 2)

That source line is Just Wrong. Use the constant os.SEEK_END rather than an obscure 2. Similarly for 0 Magic Number.

            position = source.tell()
            source.seek(0, 2)
            if position != source.tell():

Sure, I'll grant you there's a race here, the if might possibly trigger. But I think you wanted to assign position before the while loop. (Or maybe you wanted to re-open source?)

    BUFF_SIZE = 2 ** 20

That's the 2nd time you've defined that. DRY. Declare it in a common place that both classes can pull the value from.

    while header != -1 and (header > 127 or header & 3):

This is obscure. Yes, I understand the high bit distinguishes between file / dir, and the two low order bits have been defined elsewhere. That doesn't make 127 any less of a Magic Number, nor 3. You probably want to define a helper function that knows such encoding details.

        if header < 128:

Ok, if you're going to hard code Magic Numbers, at least pick one and stick with it. Consider parsing out the high bit into its own variable, then use that. I imagine the same helper would also return type_code.

    def EOF(self):

PEP-8 asks that you name this eof(), as it is not a class. Similarly for self._eof.

    path = _os.path.join(self.__parents[_int(self.__read((header >> 4 & 7) + 1))], self.__read(_int(self.__read((header >> 3 & 1) + 1))))

I imagine this is some very nice C code, but in python land we encourage breaking out the occasional helper function.

    if header >> 2 & 1:

Again, this is obscure, despite the very nice FORMAT_DOCUMENTATION you helpfully offered. Break out a helper function with a descriptive name.

You apparently wrote some additional code in a separate code block, but this review has gone on overlong already.

Summary: code is obscure, and needs unit tests added so we can confidently re-factor.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.