Skip to content
Commits on Source (4)
......@@ -17,6 +17,4 @@ deploy:
tags: true
provider: pypi
distributions: "sdist bdist_wheel"
user: seang
password:
secure: "dB3c7Ha9wYvdbpFn/FOb1sIDI0N/qvU5RKRFSLMsgp7rPuD0Vt4T8GMMWeiN+NmbgubAbe1sFhUUzXjh6Y7y/5Lolbd7lUTJLp4G+8v27ES6/9rVjMOZwoJFeRLOzF9Sl/ZONPo7zyI/fQS7x1BXfVaJKUhnSassyPABDU9dxw8="
user: mapboxci
Changes
=======
0.5.0 (2018-09-26)
------------------
- Version 0.4.0 is not compatible with click 7, but version 0.5.0 is (#23).
- The features_in_arg handler now takes sequences of geometry objects (#14).
- The iter_features function has a new per-feature callback and is ready for
use in other projects like Fiona (#15).
- The plugins module has been removed (#17).
0.4.0 (2015-12-17)
------------------
- Introduces a click argument, `features_in_arg`, which utilizes a click
callback to normalize the input of geojson features (#9).
- Release from tagged Travis CI builds (#10).
0.3.0 (2015-08-12)
------------------
- Deprecation of the cligj.plugins module (#6). Please switch to the
click-plugins module: https://github.com/click-contrib/click-plugins. The
cligj.plugins module will be removed from cligj at version 1.0.
0.2.0 (2015-05-28)
------------------
- Addition of a pluggable command group class and a corresponding click-style
decorator (#2, #3).
0.1.0 (2015-01-06)
------------------
- Initial release: a collection of GeoJSON-related command line arguments and
options for use with Click (#1).
......@@ -101,7 +101,7 @@ a delimiter, use the ``--rs`` option
if sequence:
for feature in process_features(features):
if use_rs:
click.echo(b'\x1e', nl=False)
click.echo(u'\x1e', nl=False)
click.echo(json.dumps(feature))
else:
click.echo(json.dumps(
......@@ -148,12 +148,3 @@ And can be used like this
^^{'type': 'Feature', 'id': '2'}
In this example, ``^^`` represents 0x1e.
Plugins
-------
.. warning::
The cligj.plugins module is deprecated and will be removed at version 1.0.
Use `click-plugins <https://github.com/click-contrib/click-plugins>`_
instead.
......@@ -16,6 +16,7 @@ files_in_arg = click.argument(
required=True,
metavar="INPUTS...")
# Multiple files, last of which is an output file.
files_inout_arg = click.argument(
'files',
......@@ -24,9 +25,10 @@ files_inout_arg = click.argument(
required=True,
metavar="INPUTS... OUTPUT")
# Features input
# Accepts multiple representations of GeoJSON features
# Returns the input data as an iterable of GeoJSON Feature-like dictionaries
# Features from files, command line args, or stdin.
# Returns the input data as an iterable of GeoJSON Feature-like
# dictionaries.
features_in_arg = click.argument(
'features',
nargs=-1,
......@@ -47,7 +49,7 @@ quiet_opt = click.option(
# Format driver option.
format_opt = click.option(
'-f', '--format', '--driver',
'-f', '--format', '--driver', 'driver',
default='GTiff',
help="Output format driver")
......
......@@ -5,71 +5,151 @@ import re
import click
def normalize_feature_inputs(ctx, param, features_like):
""" Click callback which accepts the following values:
* Path to file(s), each containing single FeatureCollection or Feature
* Coordinate pair(s) of the form "[0, 0]" or "0, 0" or "0 0"
* if not specified or '-', process STDIN stream containing
- line-delimited features
- ASCII Record Separator (0x1e) delimited features
- FeatureCollection or Feature object
and yields GeoJSON Features.
"""
if len(features_like) == 0:
features_like = ('-',)
def normalize_feature_inputs(ctx, param, value):
"""Click callback that normalizes feature input values.
Returns a generator over features from the input value.
Parameters
----------
ctx: a Click context
param: the name of the argument or option
value: object
The value argument may be one of the following:
1. A list of paths to files containing GeoJSON feature
collections or feature sequences.
2. A list of string-encoded coordinate pairs of the form
"[lng, lat]", or "lng, lat", or "lng lat".
for flike in features_like:
If no value is provided, features will be read from stdin.
"""
for feature_like in value or ('-',):
try:
# It's a file/stream with GeoJSON
src = iter(click.open_file(flike, mode='r'))
for feature in iter_features(src):
yield feature
with click.open_file(feature_like) as src:
for feature in iter_features(iter(src)):
yield feature
except IOError:
# It's a coordinate string
coords = list(coords_from_query(flike))
feature = {
coords = list(coords_from_query(feature_like))
yield {
'type': 'Feature',
'properties': {},
'geometry': {
'type': 'Point',
'coordinates': coords}}
yield feature
def iter_features(src):
"""Yield features from a src that may be either a GeoJSON feature
text sequence or GeoJSON feature collection."""
first_line = next(src)
# If input is RS-delimited JSON sequence.
def iter_features(geojsonfile, func=None):
"""Extract GeoJSON features from a text file object.
Given a file-like object containing a single GeoJSON feature
collection text or a sequence of GeoJSON features, iter_features()
iterates over lines of the file and yields GeoJSON features.
Parameters
----------
geojsonfile: a file-like object
The geojsonfile implements the iterator protocol and yields
lines of JSON text.
func: function, optional
A function that will be applied to each extracted feature. It
takes a feature object and may return a replacement feature or
None -- in which case iter_features does not yield.
"""
func = func or (lambda x: x)
first_line = next(geojsonfile)
# Does the geojsonfile contain RS-delimited JSON sequences?
if first_line.startswith(u'\x1e'):
buffer = first_line.strip(u'\x1e')
for line in src:
text_buffer = first_line.strip(u'\x1e')
for line in geojsonfile:
if line.startswith(u'\x1e'):
if buffer:
feat = json.loads(buffer)
yield feat
buffer = line.strip(u'\x1e')
if text_buffer:
obj = json.loads(text_buffer)
if 'coordinates' in obj:
obj = to_feature(obj)
newfeat = func(obj)
if newfeat:
yield newfeat
text_buffer = line.strip(u'\x1e')
else:
buffer += line
text_buffer += line
# complete our parsing with a for-else clause.
else:
feat = json.loads(buffer)
yield feat
obj = json.loads(text_buffer)
if 'coordinates' in obj:
obj = to_feature(obj)
newfeat = func(obj)
if newfeat:
yield newfeat
# If not, it may contains LF-delimited GeoJSON objects or a single
# multi-line pretty-printed GeoJSON object.
else:
# Try to parse LF-delimited sequences of features or feature
# collections produced by, e.g., `jq -c ...`.
try:
feat = json.loads(first_line)
assert feat['type'] == 'Feature'
yield feat
for line in src:
feat = json.loads(line)
yield feat
except (TypeError, KeyError, AssertionError, ValueError):
text = "".join(chain([first_line], src))
feats = json.loads(text)
if feats['type'] == 'Feature':
yield feats
elif feats['type'] == 'FeatureCollection':
for feat in feats['features']:
yield feat
obj = json.loads(first_line)
if obj['type'] == 'Feature':
newfeat = func(obj)
if newfeat:
yield newfeat
for line in geojsonfile:
newfeat = func(json.loads(line))
if newfeat:
yield newfeat
elif obj['type'] == 'FeatureCollection':
for feat in obj['features']:
newfeat = func(feat)
if newfeat:
yield newfeat
elif 'coordinates' in obj:
newfeat = func(to_feature(obj))
if newfeat:
yield newfeat
for line in geojsonfile:
newfeat = func(to_feature(json.loads(line)))
if newfeat:
yield newfeat
# Indented or pretty-printed GeoJSON features or feature
# collections will fail out of the try clause above since
# they'll have no complete JSON object on their first line.
# To handle these, we slurp in the entire file and parse its
# text.
except ValueError:
text = "".join(chain([first_line], geojsonfile))
obj = json.loads(text)
if obj['type'] == 'Feature':
newfeat = func(obj)
if newfeat:
yield newfeat
elif obj['type'] == 'FeatureCollection':
for feat in obj['features']:
newfeat = func(feat)
if newfeat:
yield newfeat
elif 'coordinates' in obj:
newfeat = func(to_feature(obj))
if newfeat:
yield newfeat
def to_feature(obj):
"""Takes a feature or a geometry
returns feature verbatim or
wraps geom in a feature with empty properties
"""
if obj['type'] == 'Feature':
return obj
elif 'coordinates' in obj:
return {
'type': 'Feature',
'properties': {},
'geometry': obj}
else:
raise ValueError("Object is not a feature or geometry")
def iter_query(query):
"""Accept a filename, stream, or string.
......@@ -86,7 +166,8 @@ def coords_from_query(query):
try:
coords = json.loads(query)
except ValueError:
vals = re.split(r"\,*\s*", query.strip())
query = query.replace(',', ' ')
vals = query.split()
coords = [float(v) for v in vals]
return tuple(coords[:2])
......
"""
Common components required to enable setuptools plugins.
In general the components defined here are slightly modified or subclassed
versions of core click components. This is required in order to insert code
that loads entry points when necessary while still maintaining a simple API
is only slightly different from the click API. Here's how it works:
When defining a main commandline group:
>>> import click
>>> @click.group()
... def cli():
... '''A commandline interface.'''
... pass
The `click.group()` decorator turns `cli()` into an instance of `click.Group()`.
Subsequent commands hang off of this group:
>>> @cli.command()
... @click.argument('val')
... def printer(val):
... '''Print a value.'''
... click.echo(val)
At this point the entry points, which are just instances of `click.Command()`,
can be added to the main group with:
>>> from pkg_resources import iter_entry_points
>>> for ep in iter_entry_points('module.commands'):
... cli.add_command(ep.load())
This works but its not very Pythonic, is vulnerable to typing errors, must be
manually updated if a better method is discovered, and most importantly, if an
entry point throws an exception on completely crashes the group the command is
attached to.
A better time to load the entry points is when the group they will be attached
to is instantiated. This requires slight modifications to the `click.group()`
decorator and `click.Group()` to let them load entry points as needed. If the
modified `group()` decorator is used on the same group like this:
>>> from pkg_resources import iter_entry_points
>>> import cligj.plugins
>>> @cligj.plugins.group(plugins=iter_entry_points('module.commands'))
... def cli():
... '''A commandline interface.'''
... pass
Now the entry points are loaded before the normal `click.group()` decorator
is called, except it returns a modified `Group()` so if we hang another group
off of `cli()`:
>>> @cli.group(plugins=iter_entry_points('other_module.commands'))
... def subgroup():
... '''A subgroup with more plugins'''
... pass
We can register additional plugins in a sub-group.
Catching broken plugins is done in the modified `group()` which attaches instances
of `BrokenCommand()` to the group instead of instances of `click.Command()`. The
broken commands have special help messages and override `click.Command.invoke()`
so the user gets a useful error message with a traceback if they attempt to run
the command or use `--help`.
"""
import os
import sys
import traceback
import warnings
import click
warnings.warn(
"cligj.plugins has been deprecated in favor of click-plugins: "
"https://github.com/click-contrib/click-plugins. The plugins "
"module will be removed in cligj 1.0.",
FutureWarning, stacklevel=2)
class BrokenCommand(click.Command):
"""
Rather than completely crash the CLI when a broken plugin is loaded, this
class provides a modified help message informing the user that the plugin is
broken and they should contact the owner. If the user executes the plugin
or specifies `--help` a traceback is reported showing the exception the
plugin loader encountered.
"""
def __init__(self, name):
"""
Define the special help messages after instantiating `click.Command()`.
Parameters
----------
name : str
Name of command.
"""
click.Command.__init__(self, name)
util_name = os.path.basename(sys.argv and sys.argv[0] or __file__)
if os.environ.get('CLIGJ_HONESTLY'): # pragma no cover
icon = u'\U0001F4A9'
else:
icon = u'\u2020'
self.help = (
"\nWarning: entry point could not be loaded. Contact "
"its author for help.\n\n\b\n"
+ traceback.format_exc())
self.short_help = (
icon + " Warning: could not load plugin. See `%s %s --help`."
% (util_name, self.name))
def invoke(self, ctx):
"""
Print the error message instead of doing nothing.
Parameters
----------
ctx : click.Context
Required for click.
"""
click.echo(self.help, color=ctx.color)
ctx.exit(1) # Defaults to 0 but we want an error code
class Group(click.Group):
"""
A subclass of `click.Group()` that returns the modified `group()` decorator
when `Group.group()` is called. Used by the modified `group()` decorator.
So many groups...
See the main docstring in this file for a full explanation.
"""
def __init__(self, **kwargs):
click.Group.__init__(self, **kwargs)
def group(self, *args, **kwargs):
"""
Return the modified `group()` rather than `click.group()`. This
gives the user an opportunity to assign entire groups of plugins
to their own subcommand group.
See the main docstring in this file for a full explanation.
"""
def decorator(f):
cmd = group(*args, **kwargs)(f)
self.add_command(cmd)
return cmd
return decorator
def group(plugins=None, **kwargs):
"""
A special group decorator that behaves exactly like `click.group()` but
allows for additional plugins to be loaded.
Example:
>>> import cligj.plugins
>>> from pkg_resources import iter_entry_points
>>> plugins = iter_entry_points('module.entry_points')
>>> @cligj.plugins.group(plugins=plugins)
... def cli():
... '''A CLI aplication'''
... pass
Plugins that raise an exception on load are caught and converted to an
instance of `BrokenCommand()`, which has better error handling and prevents
broken plugins from taking crashing the CLI.
See the main docstring in this file for a full explanation.
Parameters
----------
plugins : iter
An iterable that produces one entry point per iteration.
kwargs : **kwargs
Additional arguments for `click.Group()`.
"""
def decorator(f):
kwargs.setdefault('cls', Group)
grp = click.group(**kwargs)(f)
if plugins is not None:
for entry_point in plugins:
try:
grp.add_command(entry_point.load())
except Exception:
# Catch this so a busted plugin doesn't take down the CLI.
# Handled by registering a dummy command that does nothing
# other than explain the error.
grp.add_command(BrokenCommand(entry_point.name))
return grp
return decorator
python-cligj (0.4.0-3) UNRELEASED; urgency=medium
python-cligj (0.5.0-1) unstable; urgency=medium
* Team upload.
* New upstream release.
* Bump Standards-Version to 4.2.1, no changes.
* Drop autopkgtests to test installability & module import.
* Add lintian override for testsuite-autopkgtest-missing.
* Update watch file to limit matches to archive path.
-- Bas Couwenberg <sebastic@debian.org> Sun, 05 Aug 2018 20:50:24 +0200
-- Bas Couwenberg <sebastic@debian.org> Thu, 27 Sep 2018 07:18:03 +0200
python-cligj (0.4.0-2) unstable; urgency=medium
......
......@@ -8,7 +8,7 @@ with codecs_open('README.rst', encoding='utf-8') as f:
setup(name='cligj',
version='0.4.0',
version='0.5.0',
description=u"Click params for commmand line interfaces to GeoJSON",
long_description=long_description,
classifiers=[],
......@@ -21,7 +21,7 @@ setup(name='cligj',
include_package_data=True,
zip_safe=False,
install_requires=[
'click>=4.0'
'click >= 4.0, < 8'
],
extras_require={
'test': ['pytest-cov'],
......
# Do not delete this file. It makes the tests directory behave like a Python
# module, which is required to manually register and test plugins.
\ No newline at end of file
"""
We detect plugins that throw an exception on import, so just throw an exception
to mimic a problem.
"""
import click
@click.command()
def something(arg):
click.echo('passed')
raise Exception('I am a broken plugin. Send help.')
@click.command()
def after():
pass
{
"coordinates": [-122.7282, 45.5801],
"type": "Point"
}
......@@ -4,7 +4,7 @@ import sys
import pytest
from cligj.features import \
coords_from_query, iter_query, \
coords_from_query, iter_query, to_feature, \
normalize_feature_inputs, normalize_feature_objects
......@@ -46,7 +46,14 @@ def _geoms(features):
def test_featurecollection_file(expected_features):
features = normalize_feature_inputs(None, 'features', ["tests/twopoints.geojson"])
features = normalize_feature_inputs(
None, 'features', ["tests/twopoints.geojson"])
assert _geoms(features) == _geoms(expected_features)
def test_featurecollection_pretty_file(expected_features):
features = normalize_feature_inputs(
None, 'features', ["tests/twopoints-pretty.json"])
assert _geoms(features) == _geoms(expected_features)
......@@ -57,7 +64,8 @@ def test_featurecollection_stdin(expected_features):
def test_featuresequence(expected_features):
features = normalize_feature_inputs(None, 'features', ["tests/twopoints_seq.txt"])
features = normalize_feature_inputs(
None, 'features', ["tests/twopoints_seq.txt"])
assert _geoms(features) == _geoms(expected_features)
# TODO test path to sequence files fail
......@@ -69,7 +77,8 @@ def test_featuresequence_stdin(expected_features):
def test_singlefeature(expected_features):
features = normalize_feature_inputs(None, 'features', ["tests/onepoint.geojson"])
features = normalize_feature_inputs(
None, 'features', ["tests/onepoint.geojson"])
assert _geoms(features) == _geoms([expected_features[0]])
......@@ -80,7 +89,8 @@ def test_singlefeature_stdin(expected_features):
def test_featuresequencers(expected_features):
features = normalize_feature_inputs(None, 'features', ["tests/twopoints_seqrs.txt"])
features = normalize_feature_inputs(
None, 'features', ["tests/twopoints_seqrs.txt"])
assert _geoms(features) == _geoms(expected_features)
......@@ -108,6 +118,20 @@ def test_coordpairs_space(expected_features):
assert _geoms(features) == _geoms(expected_features)
def test_geometrysequence(expected_features):
features = normalize_feature_inputs(None, 'features', ["tests/twopoints_geom_seq.txt"])
assert _geoms(features) == _geoms(expected_features)
def test_geometrysequencers(expected_features):
features = normalize_feature_inputs(None, 'features', ["tests/twopoints_geom_seqrs.txt"])
assert _geoms(features) == _geoms(expected_features)
def test_geometrypretty(expected_features):
features = normalize_feature_inputs(None, 'features', ["tests/point_pretty_geom.txt"])
assert _geoms(features)[0] == _geoms(expected_features)[0]
class MockGeo(object):
def __init__(self, feature):
self.__geo_interface__ = feature
......@@ -124,3 +148,10 @@ def test_normalize_feature_objects_bad(expected_features):
objs.append(MockGeo(dict()))
with pytest.raises(ValueError):
list(normalize_feature_objects(objs))
def test_to_feature(expected_features):
geom = expected_features[0]['geometry']
feat = {'type': 'Feature', 'properties': {}, 'geometry': geom}
assert to_feature(feat) == to_feature(geom)
with pytest.raises(ValueError):
assert to_feature({'type': 'foo'})
"""Unittests for ``cligj.plugins``."""
import os
from pkg_resources import EntryPoint
from pkg_resources import iter_entry_points
from pkg_resources import working_set
import click
import cligj.plugins
# Create a few CLI commands for testing
@click.command()
@click.argument('arg')
def cmd1(arg):
"""Test command 1"""
click.echo('passed')
@click.command()
@click.argument('arg')
def cmd2(arg):
"""Test command 2"""
click.echo('passed')
# Manually register plugins in an entry point and put broken plugins in a
# different entry point.
# The `DistStub()` class gets around an exception that is raised when
# `entry_point.load()` is called. By default `load()` has `requires=True`
# which calls `dist.requires()` and the `cligj.plugins.group()` decorator
# doesn't allow us to change this. Because we are manually registering these
# plugins the `dist` attribute is `None` so we can just create a stub that
# always returns an empty list since we don't have any requirements. A full
# `pkg_resources.Distribution()` instance is not needed because there isn't
# a package installed anywhere.
class DistStub(object):
def requires(self, *args):
return []
working_set.by_key['cligj']._ep_map = {
'cligj.test_plugins': {
'cmd1': EntryPoint.parse(
'cmd1=tests.test_plugins:cmd1', dist=DistStub()),
'cmd2': EntryPoint.parse(
'cmd2=tests.test_plugins:cmd2', dist=DistStub())
},
'cligj.broken_plugins': {
'before': EntryPoint.parse(
'before=tests.broken_plugins:before', dist=DistStub()),
'after': EntryPoint.parse(
'after=tests.broken_plugins:after', dist=DistStub()),
'do_not_exist': EntryPoint.parse(
'do_not_exist=tests.broken_plugins:do_not_exist', dist=DistStub())
}
}
# Main CLI groups - one with good plugins attached and the other broken
@cligj.plugins.group(plugins=iter_entry_points('cligj.test_plugins'))
def good_cli():
"""Good CLI group."""
pass
@cligj.plugins.group(plugins=iter_entry_points('cligj.broken_plugins'))
def broken_cli():
"""Broken CLI group."""
pass
def test_registered():
# Make sure the plugins are properly registered. If this test fails it
# means that some of the for loops in other tests may not be executing.
assert len([ep for ep in iter_entry_points('cligj.test_plugins')]) > 1
assert len([ep for ep in iter_entry_points('cligj.broken_plugins')]) > 1
def test_register_and_run(runner):
result = runner.invoke(good_cli)
assert result.exit_code is 0
for ep in iter_entry_points('cligj.test_plugins'):
cmd_result = runner.invoke(good_cli, [ep.name, 'something'])
assert cmd_result.exit_code is 0
assert cmd_result.output.strip() == 'passed'
def test_broken_register_and_run(runner):
result = runner.invoke(broken_cli)
assert result.exit_code is 0
assert u'\U0001F4A9' in result.output or u'\u2020' in result.output
for ep in iter_entry_points('cligj.broken_plugins'):
cmd_result = runner.invoke(broken_cli, [ep.name])
assert cmd_result.exit_code is not 0
assert 'Traceback' in cmd_result.output
def test_group_chain(runner):
# Attach a sub-group to a CLI and get execute it without arguments to make
# sure both the sub-group and all the parent group's commands are present
@good_cli.group()
def sub_cli():
"""Sub CLI."""
pass
result = runner.invoke(good_cli)
assert result.exit_code is 0
assert sub_cli.name in result.output
for ep in iter_entry_points('cligj.test_plugins'):
assert ep.name in result.output
# Same as above but the sub-group has plugins
@good_cli.group(plugins=iter_entry_points('cligj.test_plugins'))
def sub_cli_plugins():
"""Sub CLI with plugins."""
pass
result = runner.invoke(good_cli, ['sub_cli_plugins'])
assert result.exit_code is 0
for ep in iter_entry_points('cligj.test_plugins'):
assert ep.name in result.output
# Execute one of the sub-group's commands
result = runner.invoke(good_cli, ['sub_cli_plugins', 'cmd1', 'something'])
assert result.exit_code is 0
assert result.output.strip() == 'passed'
{
"features": [
{
"bbox": [
-122.9292140099711,
45.37948199034149,
-122.44106199104115,
45.858097009742835
],
"center": [
-122.7282,
45.5801
],
"context": [
{
"id": "postcode.2503633822",
"text": "97203"
},
{
"id": "region.3470299826",
"text": "Oregon"
},
{
"id": "country.4150104525",
"short_code": "us",
"text": "United States"
}
],
"geometry": {
"coordinates": [
-122.7282,
45.5801
],
"type": "Point"
},
"id": "place.42767",
"place_name": "Portland, Oregon, United States",
"properties": {},
"relevance": 0.999,
"text": "Portland",
"type": "Feature"
},
{
"bbox": [
-121.9779540096568,
43.74737999114854,
-120.74788099000016,
44.32812500969035
],
"center": [
-121.3153,
44.0582
],
"context": [
{
"id": "postcode.3332732485",
"text": "97701"
},
{
"id": "region.3470299826",
"text": "Oregon"
},
{
"id": "country.4150104525",
"short_code": "us",
"text": "United States"
}
],
"geometry": {
"coordinates": [
-121.3153,
44.0582
],
"type": "Point"
},
"id": "place.3965",
"place_name": "Bend, Oregon, United States",
"properties": {},
"relevance": 0.999,
"text": "Bend",
"type": "Feature"
}
],
"type": "FeatureCollection"
}
{"coordinates": [-122.7282, 45.5801], "type": "Point"}
{"coordinates": [-121.3153, 44.0582], "type": "Point"}
{
"coordinates": [-122.7282, 45.5801],
"type": "Point"
}
{
"coordinates": [-121.3153, 44.0582],
"type": "Point"
}