5
\$\begingroup\$

Building on previous reviews

Whilst not an iterative review I have tried to follow the advice provided to some of my older questions:

  1. KISS, in some of my previous questions I've over complicated the design.

    Even though I'm doing quite a lot I've tried to keep everything as simple as possible. I think I achieved this in the majority of my code, where only my coroutines may have been a bit heavy handed. Even then I think the design still is fairly simple as the magic is hidden away in a fully documented wrapper.

  2. Lack of documentation / comments.

    In this program I feel that only one function required comments as it performs some magic that I didn't really understand when implementing it.

    I've also added docstrings to everything, which removes the need for most comments. I feel some of my docstrings are good and achieve their goal of saying what the program does. However the majority I feel are somewhat underwhelming, as they're pretty much just a single comment echoing the name of the function/class/method they're 'documenting'.

  3. PEP 8 violations.

    I've run a rather large amount of static analysis tools over the code. Most issues raised by these tools are, as far as I know, no longer PEP 8 issues. The majority are that mypy and pylint can't import the libraries I've used.

And so if my code exhibits any of these problems it would be great to know, with how to fix it.
Don't forget that an answer only needs to make one insightful observation.

What my program does

My full code can be seen on Git Hub and the documentation can be built using tox -e docs. Alternately you can view the documentation online. Whilst this page explains the entire code in greater depth the following is a short description.

Running segd meta.codereview does:

  1. Download the stack exchange data dump for the wanted site, caching the archive in .cache/.
  2. Extract the 7z archive, caching the files in .cache/meta.codereview.
  3. Read the XML data for comments and posts.
  4. Extract all links from the posts and comments. Using docutils to convert comments to html, and using beautiful soup to extract the hyperlinks.
  5. Save all links in meta.codereview.edges.csv and post information in meta.codereview.nodes.csv.

My code

To keep my question from becoming humungus, I have included only the code in the helpers package. These run independantly of the rest of the code, but can depend on each other.

SI

Converts a number, like 100000, into a short form, such as ' 97.66KiB'. The functions in the Magnitude class are used to convert the value to be at most the base's size. For Magnitude.ibyte this means it'd be at most 1023. display is then used to display the number to a certain amount of decimal places. This makes an easy to use library when displaying quantities of data.

"""Simplify a number to a wanted base."""

import math
from typing import Callable, Tuple


def si_magnitude(
    base: int,
    suffix: str,
    prefixes: str,
) -> Callable[[int], Tuple[int, str]]:
    """
    SI base converter builder.

    :param base: Base to truncate values to.
    :param suffix: Suffix used to denote the type of information.
    :param prefixes: Prefixes before the suffix to denote magnitude.
    :return: A function to change a value by the above parameters.
    """
    prefixes = ' '.join(prefixes.split('|')[::-1])
    prefixes_ = prefixes.split(' ')

    def inner(value: int) -> Tuple[int, str]:
        """
        Convert a number to a truncated base form.

        :param value: Value to adjust.
        :return: Truncated value and unit.
        """
        logged = math.log(value, base)
        if -1 < value < 1:
            logged -= 1
        remainder = value / base ** int(logged)
        return remainder, prefixes_[int(logged)] + suffix
    return inner


_MAGNITUDE = 'f p n μ m| k M G T P E Z Y'


class Magnitude:
    """Magnitude conversions."""

    ibyte = si_magnitude(
        1024,
        'B',
        '| Ki Mi Gi Ti Pi Ei Zi Yi',
    )
    byte = si_magnitude(
        1000,
        'B',
        _MAGNITUDE,
    )
    number = si_magnitude(
        1000,
        '',
        _MAGNITUDE,
    )


def display(values: Tuple[int, str], decimal_places: int = 2) -> str:
    """
    Display a truncated number to a wanted DP.

    :param values: Value and unit to display.
    :param decimal_places: Amount of decimal places to display the value to.
    :return: Right aligned display value.
    """
    value, unit = values
    decimal_places = int(decimal_places)
    width = 4 + decimal_places
    if decimal_places > 0:
        return f'{value:>{width}.{decimal_places}f}{unit}'
    value = int(value)
    return f'{value:>3}{unit}'

Progress

This wraps an iterator and displays how far into the iterator you are, with some extra information like total size and speed. It displays it's numbers using the SI module, so they're in a human readable form.

"""Display progress of a stream."""

import time
import warnings
from typing import Callable, Generic, Iterator, Optional, Tuple, TypeVar

from .si import Magnitude, display

# nosa(1): pylint[:Class name "T" doesn't conform to PascalCase naming style]
T = TypeVar('T')


# nosa(1): pylint[:Too many instance attributes]
class BaseProgressStream(Generic[T]):
    """Display the progress of a stream."""

    # nosa(1): pylint[:Too many arguments]
    def __init__(
        self,
        stream: Iterator[T],
        size: Optional[int],
        si: Callable[[int], Tuple[int, str]],
        progress: Callable[[T], int],
        width: int = 20,
        prefix: str = '',
        start: int = 0,
        message: Optional[str] = None,
    ):
        """Initialize BaseProgressStream."""
        self.stream = stream
        self.size = size
        self.width = width
        self.progress_bar = '=' * (width - 1) + '>'
        self.prefix = prefix
        self.to_readable = si
        self.progress_fn = progress
        self._start = start
        self.message = message

    def _get_progress(self, current: int) -> str:
        """
        Get the progress of the stream.

        :param current: Current progress - not in percentage.
        :return: Progress bar and file size.
        """
        if not self.size:
            return ''
        amount = self.width * current // self.size
        progress = self.progress_bar[-amount:] if amount else ''
        disp_size = display(self.to_readable(self.size))
        return f'[{progress:<{self.width}}] {disp_size} '

    def __iter__(self) -> Iterator[T]:
        """
        Echo the stream, and update progress.

        Catches all warnings raised whilst processing the stream to be
        displayed afterwards. This keeps the UI tidy and prevents the
        progress bar traveling over multiple lines.

        :return: An echo of the input stream.
        """
        with warnings.catch_warnings(record=True) as warnings_:
            current = self._start
            if self.message:
                print(self.message)
            start = time.clock()
            for chunk in self.stream:
                current += self.progress_fn(chunk)
                progress = self._get_progress(current)
                rate = current // max(int(time.clock() - start), 1)
                disp_rate = display(self.to_readable(rate))
                print(
                    f'\r{self.prefix}{progress}{disp_rate}/s',
                    end='',
                    flush=True,
                )
                yield chunk
            print()
        for warning in warnings_:
            warnings.showwarning(
                warning.message,
                warning.category,
                warning.filename,
                warning.lineno,
            )


class DataProgressStream(BaseProgressStream[T]):
    """Display progress of a data stream."""

    def __init__(
        self,
        stream: Iterator[T],
        size: Optional[int],
        width: int = 20,
        prefix: str = '',
        message: Optional[str] = None,
    ):
        """Initialize DataProgressStream."""
        super().__init__(
            stream,
            size,
            Magnitude.ibyte,
            len,
            width,
            prefix,
            0,
            message,
        )


class ItemProgressStream(BaseProgressStream[T]):
    """Display progress of an item stream."""

    def __init__(
        self,
        stream: Iterator[T],
        size: Optional[int],
        width: int = 20,
        prefix: str = '',
        message: Optional[str] = None,
    ):
        """Initialize ItemProgressStream."""
        super().__init__(
            stream,
            size,
            Magnitude.number,
            lambda _: 1,
            width,
            prefix,
            1,
            message,
        )

Curl

A small library that mimics curl, it displays information about the download by using the Progress module.

"""Copy URL."""

import os
import pathlib
from typing import Any

# nosa(1): pylint
import requests

from . import progress


def curl(
    path: pathlib.Path,
    *args: Any,
    **kwargs: Any,
) -> None:
    """
    Download file to system.

    Provides a progress bar of the file being downloaded and some
    statistics around the file and download.

    :param path: Local path to save the file to.
    :param args&kwargs: Passed to :code:`request.get`.
    """
    response = requests.get(*args, stream=True, **kwargs)
    response.raise_for_status()
    length_ = response.headers.get('content-length')
    length = int(length_) if length_ else None
    path.parent.mkdir(parents=True, exist_ok=True)

    print(f'Downloading: {response.url}')
    try:
        with path.open('wb') as output:
            for chunk in progress.DataProgressStream(
                response.iter_content(chunk_size=512),
                length,
                prefix='  ',
            ):
                output.write(chunk)
    except BaseException:
        os.remove(path)
        raise

Cache

Exposes a very simple cache. The objects returned all expose an ensure method to allow the action to happen at a later point, and means that a cache can rely on another cache. So the archive can rely on the file downloader if the 7z archive it needs doesn't exist.

"""
Simple file cache.

Exposes two forms of cache:

1. A file that is downloaded from a website.
2. A 7z archive cache - files that are extracted from a 7z archive.
"""

import pathlib

# nosa(1): pylint,mypy
import py7zlib

from . import curl, si


class CacheMethod:
    """Base cache object."""

    def __init__(self, cache_path: pathlib.Path) -> None:
        """Initialize CacheMethod."""
        self.cache_path = cache_path

    def _is_cached(self, use_cache: bool) -> bool:
        """
        Check if the target exist in the cache.

        :param use_cache: Set to false to force redownload the data.
        :return: True if we should use the cache.
        """
        return use_cache and self.cache_path.exists()

    def ensure(self, use_cache: bool = True) -> pathlib.Path:
        """
        Ensure target file exists.

        This should be overwritten in child classes.

        :param use_cache: Set to false to force redownload the data.
        :return: Location of file.
        """
        raise NotImplementedError('Should be overwritten in subclass.')


class FileCache(CacheMethod):
    """Exposes a cache that allows downloading files."""

    def __init__(self, cache_path: pathlib.Path, url: str) -> None:
        """Initialize FileCache."""
        super().__init__(cache_path)
        self.url = url

    def ensure(self, use_cache: bool = True) -> pathlib.Path:
        """
        Ensure target file exists.

        This curls the file from the web to cache, providing a progress
        bar whilst downloading.

        :param use_cache: Set to false to force redownload the data.
        :return: Location of file.
        """
        if not self._is_cached(use_cache):
            curl.curl(self.cache_path, self.url)
        return self.cache_path


class Archive7zCache(CacheMethod):
    """Exposes a cache that allows unzipping 7z archives."""

    def __init__(
        self,
        cache_path: pathlib.Path,
        archive_cache: CacheMethod,
    ) -> None:
        """Initialize Archive7zCache."""
        super().__init__(cache_path)
        self.archive_cache = archive_cache

    def ensure(self, use_cache: bool = True) -> pathlib.Path:
        """
        Ensure target file exists.

        Unzips the 7z archive showing the name and size of each file
        being extracted.

        :param use_cache: Set to false to force reunarchiving of the data.
        :return: Location of file.
        """
        if not self._is_cached(use_cache):
            with self.archive_cache.ensure(use_cache).open('rb') as input_file:
                print(f'Unziping: {input_file.name}')
                archive = py7zlib.Archive7z(input_file)
                directory = self.cache_path.parent
                directory.mkdir(parents=True, exist_ok=True)
                for name in archive.getnames():
                    output = directory / name
                    member = archive.getmember(name)
                    size = si.display(si.Magnitude.ibyte(member.size))
                    print(f'  Unpacking[{size}] {name}')
                    with output.open('wb') as output_file:
                        output_file.write(archive.getmember(name).read())
        return self.cache_path


class Cache:
    """Interface to make cache instances."""

    def __init__(self, cache_dir: pathlib.Path) -> None:
        """Initialize Cache."""
        self.cache_dir = cache_dir

    def file(self, cache_path: str, url: str) -> FileCache:
        """
        Get a file cache endpoint.

        :param cache_path: Location of file relative to the cache directory.
        :param url: URL location of the file to download from if not cached.
        :return: A file cache endpoint.
        """
        return FileCache(self.cache_dir / cache_path, url)

    def archive_7z(
        self,
        cache_path: pathlib.Path,
        archive_cache: CacheMethod,
    ) -> Archive7zCache:
        """
        Get an archive cache endpoint.

        :param cache_path: Location of file relative to the cache directory.
        :param archive_cache: A cache endpoint to get the 7z archive from.
        :return: An archive cache endpoint.
        """
        return Archive7zCache(self.cache_dir / cache_path, archive_cache)

XRef

This expands Sphinx partial_xref objects to be full urls. This is as you can link to posts using [link description](/q/1). Secondly Simon once didn't wrap an example link in back ticks and it makes the code blow up otherwise, so we handle that too. An example is [link description](target).

"""Expand partial xrefs."""
# nosa: pylint,mypy

from typing import List, Type

import docutils.core
import docutils.nodes
import docutils.parsers
import docutils.transforms

from recommonmark.parser import CommonMarkParser

import sphinx.addnodes

__all__ = [
    'custom_parser',
]


def custom_parser(prefix: str) -> Type[docutils.parsers.Parser]:
    """
    Markdown parser with partial xref support.

    Extends :code:`recommonmark.parser.CommonMarkParser` with to include
    the :code:`custom_parser.PendingXRefTransform` transform.

    :param prefix: Http base to prepend to partial hyperlinks.
    :return: A custom parser to parse Markdown.
    """
    class PendingXRefTransform(docutils.transforms.Transform):
        """
        Expands partial links.

        Some links are provided like :code:`[text](/a/2)``.
        This expands the link to include the basename like:

        .. :
           http://codereview.meta.stackexchange.com
        """

        default_priority = 999

        @staticmethod
        def handle_xref(
                node: sphinx.addnodes.pending_xref,
        ) -> docutils.nodes.Node:
            """Convert partial_xref to desired output."""
            referance, = node.children
            ref = node.attributes['reftarget']
            if ref != referance.attributes['refuri']:
                print(
                    'target not the same',
                    node.attributes['reftarget'],
                    referance.attributes['refuri'],
                )

            if ref.startswith('/'):
                referance['refuri'] = prefix + ref
                return referance
            # Handles 'links' like [this other thing](link)
            text, = referance.children
            if not isinstance(text, docutils.nodes.Text):
                print('Referance text is not text.')
            return docutils.nodes.Text(f'[{text.rawsource}]({ref})')

        def traverse(self, node: docutils.nodes.Node) -> None:
            """Traverse the tree updating partial_xref nodes."""
            transforms = []
            children = []
            for child in getattr(node, 'children', [])[:]:
                if isinstance(child, sphinx.addnodes.pending_xref):
                    new_child = self.handle_xref(child)
                    transforms.append((child, new_child))
                    child = new_child
                children.append(child)

            replace = getattr(node, 'replace', None)
            if replace is not None:
                for old, new in transforms:
                    replace(old, new)

            for child in children:
                self.traverse(child)

        def apply(self) -> None:
            """Docutils entry."""
            self.traverse(self.document)

    class CustomParser(CommonMarkParser):
        """Subclass of CommonMark to add XRef transform."""

        def get_transforms(self) -> List[Type[docutils.transforms.Transform]]:
            """Get transformations used for this passer."""
            return [PendingXRefTransform]

    return CustomParser

Coroutines

The main function coroutine adds a lot of magic, that's better described in the code. For the most part it doesn't do much unless the control flow enters a bad state.

"""
Coroutine helpers.

A lot of this module is based on the assumption that Python doesn't
seamlessly handle the destruction of coroutines when using multiplexing
or broadcasting. It also helps ease interactions when coroutines enter
closed states prematurely.
"""

import functools
import itertools
import types
from typing import (
    Any, Callable, Generator, Iterable, Iterator, List, Optional, Tuple, Union,
)

NEW_SOURCE = object()
EXIT = object()

IIter = Union[Iterator, Iterable]


class CoroutineDelegator:
    """Helper class for delegating to coroutines."""

    _queue: List[Tuple[IIter, Generator]]

    def __init__(self) -> None:
        """Initialize CoroutineDelegator."""
        self._queue = []

    def send_to(
        self,
        source: IIter,
        target: Generator,
    ) -> None:
        """
        Add a source and target to send data to.

        This does not send any data into the target, to do that use the
        :meth:`CoroutineDelegator.run` function.

        :param source: Input data, can be any iterable. Each is passed
                       straight unaltered to target.
        :param target: This is the coroutine the data enters into to get
                       into the coroutine control flow.
        """
        self._queue.append((source, target))

    def _increment_coroutine_refs(self) -> None:
        """Increment the amount of sources for the coroutines."""
        for _, target in self._queue:
            if _is_magic_coroutine(target):
                target.send(NEW_SOURCE)

    def _run(self, source: IIter, target: Generator) -> Optional[Iterator]:
        item = sentinel = object()
        source_ = iter(source)
        try:
            for item in source_:
                target.send(item)
        except StopIteration:
            if item is sentinel:
                return source_
            return itertools.chain([item], source_)
        else:
            if _is_magic_coroutine(target):
                target.send(EXIT)
        return None

    def run(self) -> List[Iterator]:
        """
        Send all data into the coroutine control flow.

        :return: If a coroutine is closed prematurely the data that
                 hasn't been entered into the control flow will be
                 returned. Otherwise an empty list is.
        """
        self._increment_coroutine_refs()

        output: List[Optional[Iterator]] = [
            None for _ in range(len(self._queue))
        ]
        for i, (source, target) in enumerate(self._queue):
            output[i] = self._run(source, target)
        self._queue = []
        if any(output):
            return [iter(o or []) for o in output]
        return []


def primed_coroutine(function: Callable[..., Generator]) -> Callable:
    """
    Primes a coroutine at creation.

    :param function: A coroutine function.
    :return: The coroutine function wrapped to prime the coroutine at creation.
    """
    function = types.coroutine(function)

    def inner(*args: Any, **kwargs: Any) -> Generator:
        output = function(*args, **kwargs)
        next(output)
        return output

    return inner


def _is_magic_coroutine(target: Any) -> bool:
    """
    Check if target is a magic coroutine.

    :param target: An object to check against.
    :return: If the object is a magic coroutine.
    """
    try:
        return bool(
            target
            and target.__qualname__.endswith('coroutine.<locals>.magic'),
        )
    except Exception:
        return False


def coroutine(function: Callable) -> Callable:
    """
    Wrap a coroutine generating function to make magic coroutines.

    A magic coroutine is wrapped in a protective coroutine that eases
    the destruction of coroutine pipelines. This is because the
    coroutine is wrapped in a 'bubble' that:

    1. Primes the coroutine when the first element of data is passed to it.
    2. Sends information about the creation and destruction of other
       coroutines in the pipeline. This allows a coroutine to destroy
       itself when all providers have exited.
    3. Handles when a coroutine is being prematurely closed, if this is
       the case all target coroutines will be notified that some data
       sources are no longer available allowing them to deallocate
       themselves if needed.
    4. Handles situations where a target coroutine has been prematurely
       closed. In such a situation the current coroutine will be closed
       and exit with a StopIteration error, as if the coroutine has been
       closed with the :code:`.close`.

    It should be noted that these coroutine pipelines should be started via the
    :class:`stack_exchange_graph_data.helpers.coroutines.CoroutineDelegator`.
    This is as it correctly initializes the entry coroutine, and handles
    when the coroutine has been prematurely closed.

    :param function: Standard coroutine generator function.
    :return: Function that generates magic coroutines.
    """
    self: Generator

    @primed_coroutine
    def magic(*args: Any, **kwargs: Any) -> Generator:
        # Get magic coroutine targets
        targets_ = itertools.chain(args, kwargs.values())
        targets = [
            t
            for t in targets_
            if _is_magic_coroutine(t)
        ]

        # Create wrapped coroutine
        wrapped = function(*args, **kwargs)

        # Broadcast the creation of a new source to the targets
        for target in targets:
            target.send(NEW_SOURCE)

        sources = 0
        generator_exit_flag = False
        generator_iteration_flag = False
        active = False
        try:
            # Main coroutine loop handles adding and removing source counters.
            while True:
                item = yield
                if item is NEW_SOURCE:
                    sources += 1
                elif item is EXIT:
                    sources -= 1
                    if not sources:
                        break
                else:
                    # Allows coroutines to be uninitialized until
                    # they're needed to be active.
                    if not active:
                        next(wrapped)
                        active = True
                    wrapped.send(item)
        # Raised when a anything above parent has been killed
        except RuntimeError:
            pass
        # Raised when a parent has been killed
        except StopIteration:
            generator_iteration_flag = True
        # Raised when this is being killed via `.close`.
        except GeneratorExit:
            generator_exit_flag = True
        finally:
            # Close the wrapped coroutine
            # This happens first, so any code in a `finally` can
            # propagate correctly
            try:
                wrapped.close()
            except RuntimeError:
                pass

            # Decrement target coroutine's source counters
            if targets and not generator_iteration_flag:
                for target in targets:
                    try:
                        for _ in range(sources):
                            target.send(EXIT)
                    except StopIteration:
                        pass

            # Coroutine must yield when it's being killed. IDK why but it does.
            # But it's illegal to yield when a GeneratorExit has been raised.
            if not generator_exit_flag:
                yield

    @functools.wraps(function)
    def inner(*args: Any, **kwargs: Any) -> Generator:
        nonlocal self
        self = magic(*args, **kwargs)
        return self

    return inner


@coroutine
def broadcast(*targets: Generator) -> Generator:
    """Broadcast items to targets."""
    while True:
        item = yield
        for target in targets:
            target.send(item)


@coroutine
def file_sink(*args: Any, **kwargs: Any) -> Generator:
    """Send all data to a file."""
    with open(*args, **kwargs) as file_obj:
        while True:
            file_obj.write((yield))
\$\endgroup\$
3
\$\begingroup\$

You're storing the magnitudes like this:

_MAGNITUDE = 'f p n μ m| k M G T P E Z Y'

This is a serialized format that requires parsing. This is inconvenient and complicates your code. Just store a tuple of tuples, or maybe a dictionary, storing the prefix string and its magnitude. A couple of options are:

(
  ('f', -15),
  ('p', -12),
  ('n', -9),
  # ...

(
   'mμnpf',  # Negative prefixes
   'kMGTPEZY'  # Positive prefixes
)

Otherwise... wow, this is a lot of code. It's not written terribly, but for what this is - a data-processing tool - I think it's suffering from some feature bloat. The coroutine implementation is interesting, but you don't really need this pile of code - seeing anything described as magic always gives me a sinking feeling, particularly in Python. For instance, for generic broadcasting, you can just store an iterable of function references.

\$\endgroup\$
  • 2
    \$\begingroup\$ Yeah that second way looks much better! I'm gunna have to have a think about your solution for broadcasting, my first response was "how could I be so stupid?!" but I need all the output data to perform some of the operations. When posting the question I started doubting if coroutines were the way to go, so I'll look into this more tomorrow. Thanks! \$\endgroup\$ – Peilonrayz Aug 29 at 0:48

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.