Building on previous reviews
Whilst not an iterative review I have tried to follow the advice provided to some of my older questions:
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.
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'.
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:
- Download the stack exchange data dump for the wanted site, caching the archive in
.cache/. - Extract the 7z archive, caching the files in
.cache/meta.codereview. - Read the XML data for comments and posts.
- Extract all links from the posts and comments. Using docutils to convert comments to html, and using beautiful soup to extract the hyperlinks.
- Save all links in
meta.codereview.edges.csvand post information inmeta.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))
