New Research: Supply Chain Attack on Axios Pulls Malicious Dependency from npm.Details → →
jsonquery
Artifacts are distributable versions of a package. It typically includes the source code and metadata
Advanced tools
jsonquery - pypi Package Compare versions
| 1.0.0 | ||
| ----- | ||
| * First public release |
| Metadata-Version: 1.1 | ||
| Name: jsonquery | ||
| Version: 0.1.0 | ||
| Summary: Basic json -> sqlalchemy query builder | ||
| Home-page: http://jsonquery.readthedocs.org/ | ||
| Author: Joe Cross | ||
| Author-email: joe.mcross@gmail.com | ||
| License: MIT | ||
| Description: jsonquery | ||
| ======================================================== | ||
| .. image:: https://travis-ci.org/numberoverzero/jsonquery.svg?branch=master | ||
| :target: https://travis-ci.org/numberoverzero/jsonquery | ||
| .. image:: https://coveralls.io/repos/numberoverzero/jsonquery/badge.png?branch=master | ||
| :target: https://coveralls.io/r/numberoverzero/jsonquery?branch=master | ||
| Basic json -> sqlalchemy query builder | ||
| Installation | ||
| ======================================================== | ||
| :: | ||
| pip install jsonquery | ||
| Basic Usage | ||
| ======================================================== | ||
| Let's define a model and get an engine set up:: | ||
| from sqlalchemy import Column, Integer, String, create_engine | ||
| from sqlalchemy.orm import sessionmaker | ||
| from sqlalchemy.ext.declarative import declarative_base | ||
| Base = declarative_base() | ||
| class User(Base): | ||
| __tablename__ = 'users' | ||
| id = Column(Integer, primary_key=True) | ||
| name = Column(String) | ||
| email = Column(String) | ||
| age = Column(Integer) | ||
| height = Column(Integer) | ||
| engine = create_engine("sqlite://", echo=True) | ||
| Base.metadata.create_all(engine) | ||
| model = User | ||
| session = sessionmaker(bind=engine)() | ||
| We want to get all users whose name starts with 'Pat' and are | ||
| at least 21:: | ||
| from jsonquery import jsonquery | ||
| json = { | ||
| "operator": "and", | ||
| "value": [ | ||
| { | ||
| "operator": ">=", | ||
| "column": "age", | ||
| "value": 21 | ||
| }, | ||
| { | ||
| "operator": "ilike", | ||
| "column": "name", | ||
| "value": "pat%" | ||
| } | ||
| ] | ||
| } | ||
| query = jsonquery(session, User, json) | ||
| users = query.all() | ||
| Supported Data Types | ||
| ======================================================== | ||
| jsonquery doesn't care about column type. Instead, it uses a whitelist of operators, | ||
| where keys are strings (the same that would be passed in the "operator" field of a node) | ||
| and the values are functions that take a column object and a value and return a | ||
| sqlalchemy criterion. Here are some examples:: | ||
| def greater_than(column, value): | ||
| return column > value | ||
| register_operator(">", greater_than) | ||
| def like(column, value): | ||
| like_func = getattr(column, 'like') | ||
| return like_func(value) | ||
| register_operator("like", like) | ||
| By default, the following are registered:: | ||
| >, >=, ==, !=, <=, < | ||
| like, ilike, in_ | ||
| Use ``unregister_operator(opstring)`` to remove an operator. | ||
| Future Goals | ||
| ======================================================== | ||
| There are a few features I want to add, but these are mostly convenience and aren't necessary to | ||
| the core application, which I believe is satisfied. | ||
| Compressed and/or format | ||
| -------------------------------------------------------- | ||
| Reduce repetitive column and operator specification when possible by allowing non-scalar values | ||
| for column operators. By flipping the nesting restriction on logical operators, we can omit | ||
| fields specified at the column level. This is especially prominent in string matching, | ||
| when the column and operator are the same, but we want to compare against 3+ values. | ||
| Currently:: | ||
| { | ||
| "operator": "or", | ||
| "value": [ | ||
| { | ||
| "column": "age", | ||
| "operator": "<=", | ||
| "value": 16 | ||
| }, | ||
| { | ||
| "column": "age", | ||
| "operator": ">=", | ||
| "value": 21 | ||
| }, | ||
| { | ||
| "column": "age", | ||
| "operator": "==", | ||
| "value": 18 | ||
| } | ||
| ] | ||
| } | ||
| With compressed logical operators:: | ||
| { | ||
| "column": "age" | ||
| "value": { | ||
| "operator": "or", | ||
| "value": [ | ||
| { | ||
| "operator": "<=", | ||
| "value": 16 | ||
| }, | ||
| { | ||
| "operator": ">=", | ||
| "value": 21 | ||
| }, | ||
| { | ||
| "operator": "==", | ||
| "value": 18 | ||
| } | ||
| ] | ||
| } | ||
| } | ||
| Or, when the operator is the same:: | ||
| { | ||
| "column": "name" | ||
| "operator": "like" | ||
| "value": { | ||
| "operator": "or", | ||
| "value": [ | ||
| "Bill", | ||
| "Mary", | ||
| "Steve" | ||
| ] | ||
| } | ||
| } | ||
| Motivation | ||
| ======================================================== | ||
| I want to build complex sql queries from a request body, and json is a nice way | ||
| to specify nested queries. As far as security is concerned, column/value names are passed | ||
| into a set of functions which is hardcoded, and is primarily either attribute lookup | ||
| (string functions like, ilike) or standard mathematical operators (operator.gt, for instance). | ||
| 1.0.0 | ||
| ----- | ||
| * First public release | ||
| Keywords: json sqlalchemy sql orm | ||
| Platform: any | ||
| Classifier: Development Status :: 5 - Production/Stable | ||
| Classifier: Intended Audience :: Developers | ||
| Classifier: License :: OSI Approved :: MIT License | ||
| Classifier: Operating System :: OS Independent | ||
| Classifier: Programming Language :: Python | ||
| Classifier: Programming Language :: Python :: 2 | ||
| Classifier: Programming Language :: Python :: 2.6 | ||
| Classifier: Programming Language :: Python :: 2.7 | ||
| Classifier: Programming Language :: Python :: 3 | ||
| Classifier: Programming Language :: Python :: 3.2 | ||
| Classifier: Programming Language :: Python :: 3.3 | ||
| Classifier: Programming Language :: Python :: 3.4 | ||
| Classifier: Topic :: Software Development :: Libraries | ||
| Classifier: Topic :: Software Development :: Libraries :: Python Modules |
| sqlalchemy |
| CHANGES.rst | ||
| MANIFEST.in | ||
| README.rst | ||
| jsonquery.py | ||
| setup.cfg | ||
| setup.py | ||
| jsonquery.egg-info/PKG-INFO | ||
| jsonquery.egg-info/SOURCES.txt | ||
| jsonquery.egg-info/dependency_links.txt | ||
| jsonquery.egg-info/requires.txt | ||
| jsonquery.egg-info/top_level.txt |
| jsonquery |
+221
| import operator | ||
| import collections | ||
| import sys | ||
| import sqlalchemy | ||
| PYTHON_VERSION = sys.version_info | ||
| if PYTHON_VERSION > (3, 0, 0): # pragma: no cover | ||
| # PYTHON 3k: strings == unicode | ||
| is_string = lambda s: isinstance(s, str) | ||
| else: # pragma: no cover | ||
| # PYTHON 2k: strings can be str or unicode | ||
| is_string = lambda s: isinstance(s, basestring) # flake8: noqa | ||
| DEFAULT_QUERY_CONSTRAINTS = { | ||
| 'max_breadth': None, | ||
| 'max_depth': None, | ||
| 'max_elements': 64 | ||
| } | ||
| OPERATORS = {} | ||
| def register_operator(opstring, func): | ||
| ''' | ||
| Registers a function so that the operator can be used in queries. | ||
| opstring: | ||
| The string used to reference this function in json queries | ||
| func: | ||
| Function that takes a column object | ||
| (sqlalchemy.orm.attributes.InstrumentedAttribute) | ||
| and a value and returns a criterion to be passed to | ||
| session.query(model).filter() | ||
| Example: Adding the >= operator | ||
| def gt(column, value): | ||
| return column >= value | ||
| register_operator('>=', gt) | ||
| # This can be simplified to: | ||
| import operator | ||
| register_operator('>=', operator.gt) | ||
| Example: Adding the column.in_ operator | ||
| def in_(column, value): | ||
| func = getattr(column, 'in_') | ||
| return func(value) | ||
| register_operator('in_', in_) | ||
| See http://docs.sqlalchemy.org/en/rel_0_8/orm/query.html\ | ||
| #sqlalchemy.orm.query.Query.filter. | ||
| ''' | ||
| OPERATORS[opstring] = func | ||
| binops = { | ||
| '<': operator.lt, | ||
| '<=': operator.le, | ||
| '!=': operator.ne, | ||
| '==': operator.eq, | ||
| '>=': operator.ge, | ||
| '>': operator.gt, | ||
| } | ||
| for opstring, func in binops.items(): | ||
| register_operator(opstring, func) | ||
| attr_funcs = [ | ||
| 'like', | ||
| 'ilike', | ||
| 'in_' | ||
| ] | ||
| def attr_op(op): | ||
| return lambda col, value: getattr(col, op)(value) | ||
| for opstring in attr_funcs: | ||
| register_operator(opstring, attr_op(opstring)) | ||
| def jsonquery(session, model, json, **kwargs): | ||
| ''' | ||
| Returns a query object built from the given json. | ||
| Usage: | ||
| query = jsonquery(session, model, json, query_constraints) | ||
| rows = query.all() | ||
| session: | ||
| SQLAlchemy session to build query on | ||
| model: | ||
| SQLAlchemy model to perform queries on | ||
| json: | ||
| Logical Operators | ||
| { | ||
| operator: 'and', | ||
| value: [ | ||
| OBJ1, | ||
| OBJ2, | ||
| ... | ||
| OBJN | ||
| ] | ||
| } | ||
| Columns: Numeric | ||
| { | ||
| column: 'age', | ||
| operator: '>=', | ||
| value: 18 | ||
| } | ||
| Columns: Strings | ||
| { | ||
| column: 'name', | ||
| operator: 'ilike', | ||
| value: 'pat%' | ||
| } | ||
| Logical operators 'and' and 'or' take an array, while 'not' | ||
| takes a single value. It is invalid to have a logical operator | ||
| as the value of a subquery. | ||
| Numeric operators are: | ||
| <, <=, ==, !=, >=, > | ||
| String operators are: | ||
| like case-sensitive match | ||
| ilike case-insensitive match | ||
| String wildcard character is "%", so "pat%" matches "patrick" | ||
| and "patty". Default escape character is '/' | ||
| max_breadth (Optional): | ||
| Maximum number of elements in a single and/or operator. | ||
| Default is None. | ||
| max_depth (Optional): | ||
| Maximum nested depth of a constraint. | ||
| Default is None. | ||
| max_elements (Optional): | ||
| Maximum number of constraints and logical operators allowed in a query. | ||
| Default is 64. | ||
| ''' | ||
| constraints = dict(DEFAULT_QUERY_CONSTRAINTS) | ||
| constraints.update(kwargs) | ||
| count = depth = 0 | ||
| criterion, total_elements = _build(json, count, depth, model, constraints) | ||
| return session.query(model).filter(criterion) | ||
| def _build(node, count, depth, model, constraints): | ||
| count += 1 | ||
| depth += 1 | ||
| value = node['value'] | ||
| _validate_query_constraints(value, count, depth, constraints) | ||
| logical_operators = { | ||
| 'and': (_build_sql_sequence, sqlalchemy.and_), | ||
| 'or': (_build_sql_sequence, sqlalchemy.or_), | ||
| 'not': (_build_sql_unary, sqlalchemy.not_), | ||
| } | ||
| op = node['operator'] | ||
| if op in logical_operators: | ||
| builder, func = logical_operators[op] | ||
| return builder(node, count, depth, model, constraints, func) | ||
| else: | ||
| return _build_column(node, model), count | ||
| def _validate_query_constraints(value, count, depth, constraints): | ||
| '''Raises if any query constraints are violated''' | ||
| max_breadth = constraints['max_breadth'] | ||
| max_depth = constraints['max_depth'] | ||
| max_elements = constraints['max_elements'] | ||
| if max_depth and depth > max_depth: | ||
| raise ValueError('Depth limit ({}) exceeded'.format(max_depth)) | ||
| element_breadth = 1 | ||
| if isinstance(value, collections.Sequence) and not is_string(value): | ||
| element_breadth = len(value) | ||
| if max_breadth and element_breadth > max_breadth: | ||
| raise ValueError( | ||
| 'Breadth limit ({}) exceeded'.format(max_breadth)) | ||
| count += element_breadth | ||
| if max_elements and count > max_elements: | ||
| raise ValueError( | ||
| 'Filter elements limit ({}) exceeded'.format(max_elements)) | ||
| def _build_sql_sequence(node, count, depth, model, constraints, func): | ||
| ''' | ||
| func is either sqlalchemy.and_ or sqlalchemy.or_ | ||
| Build each subquery in node['value'], then combine with func(*subqueries) | ||
| ''' | ||
| subqueries = [] | ||
| for value in node['value']: | ||
| subquery, count = _build(value, count, depth, model, constraints) | ||
| subqueries.append(subquery) | ||
| return func(*subqueries), count | ||
| def _build_sql_unary(node, count, depth, model, constraints, func): | ||
| ''' | ||
| func is sqlalchemy.not_ (may support others) | ||
| ''' | ||
| value = node['value'] | ||
| subquery, count = _build(value, count, depth, model, constraints) | ||
| return func(subquery), count | ||
| def _build_column(node, model): | ||
| # string => sqlalchemy.orm.attributes.InstrumentedAttribute | ||
| column = node['column'] | ||
| column = getattr(model, column) | ||
| op = node['operator'] | ||
| value = node['value'] | ||
| return OPERATORS[op](column, value) |
| include CHANGES.rst | ||
| include README.rst | ||
| recursive-exclude tests * |
+14
| [run] | ||
| omit = test* | ||
| [flake8] | ||
| exclude = .tox, dist, doc, build, *.egg | ||
| [wheel] | ||
| universal = 1 | ||
| [egg_info] | ||
| tag_build = | ||
| tag_svn_revision = 0 | ||
| tag_date = 0 | ||
+30
-6
| Metadata-Version: 1.1 | ||
| Name: jsonquery | ||
| Version: 0.2.1 | ||
| Summary: Lightweight bit packing for classes | ||
| Home-page: https://github.com/numberoverzero/jsonquery | ||
| Version: 0.1.0 | ||
| Summary: Basic json -> sqlalchemy query builder | ||
| Home-page: http://jsonquery.readthedocs.org/ | ||
| Author: Joe Cross | ||
| Author-email: joe.mcross@gmail.com | ||
| License: LICENSE.txt | ||
| License: MIT | ||
| Description: jsonquery | ||
| ======================================================== | ||
| .. image:: https://travis-ci.org/numberoverzero/jsonquery.svg?branch=master | ||
| :target: https://travis-ci.org/numberoverzero/jsonquery | ||
| .. image:: https://coveralls.io/repos/numberoverzero/jsonquery/badge.png?branch=master | ||
| :target: https://coveralls.io/r/numberoverzero/jsonquery?branch=master | ||
| Basic json -> sqlalchemy query builder | ||
@@ -176,3 +181,22 @@ | ||
| Platform: UNKNOWN | ||
| Requires: SQLAlchemy | ||
| 1.0.0 | ||
| ----- | ||
| * First public release | ||
| Keywords: json sqlalchemy sql orm | ||
| Platform: any | ||
| Classifier: Development Status :: 5 - Production/Stable | ||
| Classifier: Intended Audience :: Developers | ||
| Classifier: License :: OSI Approved :: MIT License | ||
| Classifier: Operating System :: OS Independent | ||
| Classifier: Programming Language :: Python | ||
| Classifier: Programming Language :: Python :: 2 | ||
| Classifier: Programming Language :: Python :: 2.6 | ||
| Classifier: Programming Language :: Python :: 2.7 | ||
| Classifier: Programming Language :: Python :: 3 | ||
| Classifier: Programming Language :: Python :: 3.2 | ||
| Classifier: Programming Language :: Python :: 3.3 | ||
| Classifier: Programming Language :: Python :: 3.4 | ||
| Classifier: Topic :: Software Development :: Libraries | ||
| Classifier: Topic :: Software Development :: Libraries :: Python Modules |
+5
-0
| jsonquery | ||
| ======================================================== | ||
| .. image:: https://travis-ci.org/numberoverzero/jsonquery.svg?branch=master | ||
| :target: https://travis-ci.org/numberoverzero/jsonquery | ||
| .. image:: https://coveralls.io/repos/numberoverzero/jsonquery/badge.png?branch=master | ||
| :target: https://coveralls.io/r/numberoverzero/jsonquery?branch=master | ||
| Basic json -> sqlalchemy query builder | ||
@@ -5,0 +10,0 @@ |
+52
-14
@@ -1,15 +0,53 @@ | ||
| from distutils.core import setup | ||
| """ Setup file """ | ||
| import os | ||
| import re | ||
| from setuptools import setup, find_packages | ||
| setup( | ||
| name='jsonquery', | ||
| version='0.2.1', | ||
| author='Joe Cross', | ||
| author_email='joe.mcross@gmail.com', | ||
| packages=['jsonquery'], | ||
| url='https://github.com/numberoverzero/jsonquery', | ||
| license='LICENSE.txt', | ||
| description='Lightweight bit packing for classes', | ||
| long_description=open('README.rst').read(), | ||
| install_requires=["SQLAlchemy >= 0.8.2"], | ||
| requires=["SQLAlchemy"], | ||
| ) | ||
| HERE = os.path.abspath(os.path.dirname(__file__)) | ||
| README = open(os.path.join(HERE, 'README.rst')).read() | ||
| CHANGES = open(os.path.join(HERE, 'CHANGES.rst')).read() | ||
| # Remove custom RST extensions for pypi | ||
| CHANGES = re.sub(r'\(\s*:(issue|pr|sha):.*?\)', '', CHANGES) | ||
| REQUIREMENTS = [ | ||
| 'sqlalchemy' | ||
| ] | ||
| TEST_REQUIREMENTS = [ | ||
| 'pytest' | ||
| ] | ||
| if __name__ == "__main__": | ||
| setup( | ||
| name='jsonquery', | ||
| version='0.1.0', | ||
| description="Basic json -> sqlalchemy query builder", | ||
| long_description=README + '\n\n' + CHANGES, | ||
| classifiers=[ | ||
| 'Development Status :: 5 - Production/Stable', | ||
| 'Intended Audience :: Developers', | ||
| 'License :: OSI Approved :: MIT License', | ||
| 'Operating System :: OS Independent', | ||
| 'Programming Language :: Python', | ||
| 'Programming Language :: Python :: 2', | ||
| 'Programming Language :: Python :: 2.6', | ||
| 'Programming Language :: Python :: 2.7', | ||
| 'Programming Language :: Python :: 3', | ||
| 'Programming Language :: Python :: 3.2', | ||
| 'Programming Language :: Python :: 3.3', | ||
| 'Programming Language :: Python :: 3.4', | ||
| 'Topic :: Software Development :: Libraries', | ||
| 'Topic :: Software Development :: Libraries :: Python Modules' | ||
| ], | ||
| author='Joe Cross', | ||
| author_email='joe.mcross@gmail.com', | ||
| url='http://jsonquery.readthedocs.org/', | ||
| license='MIT', | ||
| keywords='json sqlalchemy sql orm', | ||
| platforms='any', | ||
| include_package_data=True, | ||
| py_modules=['jsonquery'], | ||
| packages=find_packages(exclude=('tests',)), | ||
| install_requires=REQUIREMENTS, | ||
| tests_require=REQUIREMENTS + TEST_REQUIREMENTS, | ||
| ) |
| import operator | ||
| import collections | ||
| import sqlalchemy | ||
| DEFAULT_QUERY_CONSTRAINTS = { | ||
| 'max_breadth': None, | ||
| 'max_depth': None, | ||
| 'max_elements': 64 | ||
| } | ||
| OPERATORS = {} | ||
| def register_operator(opstring, func): | ||
| ''' | ||
| Registers a function so that the operator can be used in queries. | ||
| opstring: | ||
| The string used to reference this function in json queries | ||
| func: | ||
| Function that takes a column object (sqlalchemy.orm.attributes.InstrumentedAttribute) | ||
| and a value and returns a criterion to be passed to session.query(model).filter() | ||
| Example: Adding the >= operator | ||
| def gt(column, value): | ||
| return column >= value | ||
| register_operator('>=', gt) | ||
| # This can be simplified to: | ||
| import operator | ||
| register_operator('>=', operator.gt) | ||
| Example: Adding the column.in_ operator | ||
| def in_(column, value): | ||
| func = getattr(column, 'in_') | ||
| return func(value) | ||
| register_operator('in_', in_) | ||
| See http://docs.sqlalchemy.org/en/rel_0_8/orm/query.html#sqlalchemy.orm.query.Query.filter. | ||
| ''' | ||
| OPERATORS[opstring] = func | ||
| binops = { | ||
| '<': operator.lt, | ||
| '<=': operator.le, | ||
| '!=': operator.ne, | ||
| '==': operator.eq, | ||
| '>=': operator.ge, | ||
| '>': operator.gt, | ||
| } | ||
| for opstring, func in binops.iteritems(): | ||
| register_operator(opstring, func) | ||
| attr_funcs = [ | ||
| 'like', | ||
| 'ilike', | ||
| 'in_' | ||
| ] | ||
| def attr_op(op): return lambda col, value: getattr(col, op)(value) | ||
| for opstring in attr_funcs: | ||
| register_operator(opstring, attr_op(opstring)) | ||
| def jsonquery(session, model, json, **kwargs): | ||
| ''' | ||
| Returns a query object built from the given json. | ||
| Usage: | ||
| query = jsonquery(session, model, json, query_constraints) | ||
| rows = query.all() | ||
| session: | ||
| SQLAlchemy session to build query on | ||
| model: | ||
| SQLAlchemy model to perform queries on | ||
| json: | ||
| Logical Operators | ||
| { | ||
| operator: 'and', | ||
| value: [ | ||
| OBJ1, | ||
| OBJ2, | ||
| ... | ||
| OBJN | ||
| ] | ||
| } | ||
| Columns: Numeric | ||
| { | ||
| column: 'age', | ||
| operator: '>=', | ||
| value: 18 | ||
| } | ||
| Columns: Strings | ||
| { | ||
| column: 'name', | ||
| operator: 'ilike', | ||
| value: 'pat%' | ||
| } | ||
| Logical operators 'and' and 'or' take an array, while 'not' takes a single value. | ||
| It is invalid to have a logical operator as the value of a subquery. | ||
| Numeric operators are: | ||
| <, <=, ==, !=, >=, > | ||
| String operators are: | ||
| like case-sensitive match | ||
| ilike case-insensitive match | ||
| String wildcard character is % (so "pat%" matches "patrick" and "patty") | ||
| with default escape character '/' | ||
| max_breadth (Optional): | ||
| Maximum number of elements in a single and/or operator. Default is None. | ||
| max_depth (Optional): | ||
| Maximum nested depth of a constraint. Default is None. | ||
| max_elements (Optional): | ||
| Maximum number of constraints and logical operators allowed in a query. Default is 64. | ||
| ''' | ||
| constraints = dict(DEFAULT_QUERY_CONSTRAINTS) | ||
| constraints.update(kwargs) | ||
| count = depth = 0 | ||
| criterion, total_elements = _build(json, count, depth, model, constraints) | ||
| return session.query(model).filter(criterion) | ||
| def _build(node, count, depth, model, constraints): | ||
| count += 1 | ||
| depth += 1 | ||
| value = node['value'] | ||
| _validate_query_constraints(value, count, depth, constraints) | ||
| logical_operators = { | ||
| 'and': (_build_sql_sequence, sqlalchemy.and_), | ||
| 'or': (_build_sql_sequence, sqlalchemy.or_), | ||
| 'not': (_build_sql_unary, sqlalchemy.not_), | ||
| } | ||
| op = node['operator'] | ||
| if op in logical_operators: | ||
| builder, func = logical_operators[op] | ||
| return builder(node, count, depth, model, constraints, func) | ||
| else: | ||
| return _build_column(node, model), count | ||
| def _validate_query_constraints(value, count, depth, constraints): | ||
| '''Raises if any query constraints are violated''' | ||
| max_breadth = constraints['max_breadth'] | ||
| max_depth = constraints['max_depth'] | ||
| max_elements = constraints['max_elements'] | ||
| if max_depth and depth > max_depth: | ||
| raise ValueError('Depth limit ({}) exceeded'.format(max_depth)) | ||
| element_breadth = 1 | ||
| if isinstance(value, collections.Sequence) and not isinstance(value, basestring): | ||
| element_breadth = len(value) | ||
| if max_breadth and element_breadth > max_breadth: | ||
| raise ValueError('Breadth limit ({}) exceeded'.format(max_breadth)) | ||
| count += element_breadth | ||
| if max_elements and count > max_elements: | ||
| raise ValueError('Filter elements limit ({}) exceeded'.format(max_elements)) | ||
| def _build_sql_sequence( node, count, depth, model, constraints, func): | ||
| ''' | ||
| func is either sqlalchemy.and_ or sqlalchemy.or_ | ||
| Build each subquery in node['value'], then combine with func(*subqueries) | ||
| ''' | ||
| subqueries = [] | ||
| for value in node['value']: | ||
| subquery, count = _build(value, count, depth, model, constraints) | ||
| subqueries.append(subquery) | ||
| return func(*subqueries), count | ||
| def _build_sql_unary( node, count, depth, model, constraints, func): | ||
| ''' | ||
| func is sqlalchemy.not_ (may support others) | ||
| ''' | ||
| value = node['value'] | ||
| subquery, count = _build(value, count, depth, model, constraints) | ||
| return func(subquery), count | ||
| def _build_column(node, model): | ||
| # string => sqlalchemy.orm.attributes.InstrumentedAttribute | ||
| column = node['column'] | ||
| column = getattr(model, column) | ||
| op = node['operator'] | ||
| value = node['value'] | ||
| return OPERATORS[op](column, value) |
| import json | ||
| import unittest | ||
| from sqlalchemy import Column, Integer, String, create_engine, and_, or_, not_ | ||
| from sqlalchemy.orm import sessionmaker | ||
| from sqlalchemy.ext.declarative import declarative_base | ||
| from jsonquery import jsonquery | ||
| def jsonify(dict): | ||
| # Easy validation that the test data isn't invalid json | ||
| return json.loads(json.dumps(dict)) | ||
| class JsonQueryTestCase(unittest.TestCase): | ||
| def setUp(self): | ||
| Base = declarative_base() | ||
| class User(Base): | ||
| __tablename__ = 'users' | ||
| id = Column(Integer, primary_key=True) | ||
| name = Column(String) | ||
| email = Column(String) | ||
| age = Column(Integer) | ||
| height = Column(Integer) | ||
| engine = create_engine("sqlite://", echo=True) | ||
| Base.metadata.create_all(engine) | ||
| self.model = User | ||
| self.session = sessionmaker(bind=engine)() | ||
| def tearDown(self): | ||
| self.session.close() | ||
| def add_user(self, **kwargs): | ||
| user = self.model(**kwargs) | ||
| self.session.add(user) | ||
| self.session.commit() | ||
| @property | ||
| def query(self): | ||
| return self.session.query(self.model) | ||
| def test_basic_query(self): | ||
| self.add_user(age=10) | ||
| json = jsonify({ | ||
| 'column': 'age', | ||
| 'value': 10, | ||
| 'operator': '==' | ||
| }) | ||
| actual_user = jsonquery(self.session, self.model, json).one() | ||
| expected_user = self.query.filter(self.model.age == 10).one() | ||
| assert actual_user is expected_user | ||
| def test_element_limit(self): | ||
| self.add_user(age=10) | ||
| json = jsonify({ | ||
| 'operator': 'and', | ||
| 'value': [ | ||
| { | ||
| 'column': 'age', | ||
| 'value': 10, | ||
| 'operator': '==' | ||
| } | ||
| ] | ||
| }) | ||
| with self.assertRaises(ValueError): | ||
| jsonquery(self.session, self.model, json, max_elements=1).one() | ||
| def test_depth_limit(self): | ||
| self.add_user(age=10) | ||
| json = jsonify({ | ||
| 'operator': 'and', | ||
| 'value': [ | ||
| { | ||
| 'column': 'age', | ||
| 'value': 10, | ||
| 'operator': '==' | ||
| } | ||
| ] | ||
| }) | ||
| with self.assertRaises(ValueError): | ||
| jsonquery(self.session, self.model, json, max_depth=1).one() | ||
| def test_breadth_limit(self): | ||
| self.add_user(age=10) | ||
| json = jsonify({ | ||
| 'operator': 'and', | ||
| 'value': [ | ||
| { | ||
| 'column': 'age', | ||
| 'value': 10, | ||
| 'operator': '==' | ||
| }, | ||
| { | ||
| 'column': 'age', | ||
| 'value': 10, | ||
| 'operator': '==' | ||
| } | ||
| ] | ||
| }) | ||
| with self.assertRaises(ValueError): | ||
| jsonquery(self.session, self.model, json, max_breadth=1).one() | ||
| def test_basic_and(self): | ||
| self.add_user(age=10) | ||
| json = jsonify({ | ||
| 'operator': 'and', | ||
| 'value': [ | ||
| { | ||
| 'column': 'age', | ||
| 'value': 10, | ||
| 'operator': '==' | ||
| } | ||
| ] | ||
| }) | ||
| actual_user = jsonquery(self.session, self.model, json).one() | ||
| expected_user = self.query.filter(and_(self.model.age == 10)).one() | ||
| assert actual_user is expected_user | ||
| def test_multi_criteria_and(self): | ||
| self.add_user(age=10, height=20) | ||
| self.add_user(age=10, height=15) | ||
| self.add_user(age=5, height=15) | ||
| json = jsonify({ | ||
| 'operator': 'and', | ||
| 'value': [ | ||
| { | ||
| 'column': 'age', | ||
| 'value': 10, | ||
| 'operator': '==' | ||
| }, | ||
| { | ||
| 'column': 'height', | ||
| 'value': 15, | ||
| 'operator': '==' | ||
| } | ||
| ] | ||
| }) | ||
| actual_user = jsonquery(self.session, self.model, json).one() | ||
| expected_user = self.query.filter(and_(self.model.age == 10, self.model.height == 15)).one() | ||
| assert actual_user is expected_user | ||
| def test_basic_or(self): | ||
| self.add_user(age=10) | ||
| json = jsonify({ | ||
| 'operator': 'or', | ||
| 'value': [ | ||
| { | ||
| 'column': 'age', | ||
| 'value': 10, | ||
| 'operator': '==' | ||
| } | ||
| ] | ||
| }) | ||
| actual_user = jsonquery(self.session, self.model, json).one() | ||
| expected_user = self.query.filter(or_(self.model.age == 10)).one() | ||
| assert actual_user is expected_user | ||
| def test_multi_criteria_or(self): | ||
| self.add_user(age=10, height=20) | ||
| self.add_user(age=10, height=15) | ||
| self.add_user(age=5, height=15) | ||
| json = jsonify({ | ||
| 'operator': 'or', | ||
| 'value': [ | ||
| { | ||
| 'column': 'age', | ||
| 'value': 10, | ||
| 'operator': '==' | ||
| }, | ||
| { | ||
| 'column': 'height', | ||
| 'value': 15, | ||
| 'operator': '==' | ||
| } | ||
| ] | ||
| }) | ||
| actual_users = jsonquery(self.session, self.model, json).all() | ||
| expected_users = self.query.filter(or_(self.model.age == 10, self.model.height == 15)).all() | ||
| assert 3 == len(actual_users) == len(expected_users) | ||
| assert set(actual_users) == set(expected_users) | ||
| def test_basic_not(self): | ||
| self.add_user(age=10) | ||
| self.add_user(age=20) | ||
| self.add_user(age=30) | ||
| json = jsonify({ | ||
| 'operator': 'not', | ||
| 'value': | ||
| { | ||
| 'column': 'age', | ||
| 'value': 10, | ||
| 'operator': '==' | ||
| } | ||
| }) | ||
| actual_users = jsonquery(self.session, self.model, json).all() | ||
| expected_users = self.query.filter(not_(self.model.age == 10)).all() | ||
| assert 2 == len(actual_users) == len(expected_users) | ||
| assert set(actual_users) == set(expected_users) | ||
| class IntegerColumnTestCase(unittest.TestCase): | ||
| def setUp(self): | ||
| Base = declarative_base() | ||
| class Foo(Base): | ||
| __tablename__ = 'foos' | ||
| id = Column(Integer, primary_key=True) | ||
| foo = Column(Integer) | ||
| engine = create_engine("sqlite://", echo=True) | ||
| Base.metadata.create_all(engine) | ||
| self.model = Foo | ||
| self.session = sessionmaker(bind=engine)() | ||
| self.add_foo(10) | ||
| self.add_foo(15) | ||
| self.add_foo(20) | ||
| def tearDown(self): | ||
| self.session.close() | ||
| def add_foo(self, value): | ||
| foo = self.model(foo=value) | ||
| self.session.add(foo) | ||
| self.session.commit() | ||
| @property | ||
| def query(self): | ||
| return self.session.query(self.model) | ||
| def test_eq(self): | ||
| json = jsonify({ | ||
| 'column': 'foo', | ||
| 'value': 10, | ||
| 'operator': '==' | ||
| }) | ||
| actual_foo = jsonquery(self.session, self.model, json).one() | ||
| expected_foo = self.query.filter(self.model.foo == 10).one() | ||
| assert actual_foo is expected_foo | ||
| def test_ne(self): | ||
| json = jsonify({ | ||
| 'column': 'foo', | ||
| 'value': 10, | ||
| 'operator': '!=' | ||
| }) | ||
| actual_foos = jsonquery(self.session, self.model, json).all() | ||
| expected_foos = self.query.filter(self.model.foo != 10).all() | ||
| assert 2 == len(actual_foos) == len(expected_foos) | ||
| assert set(actual_foos) == set(expected_foos) | ||
| def test_lt(self): | ||
| json = jsonify({ | ||
| 'column': 'foo', | ||
| 'value': 15, | ||
| 'operator': '<' | ||
| }) | ||
| actual_foo = jsonquery(self.session, self.model, json).one() | ||
| expected_foo = self.query.filter(self.model.foo < 15).one() | ||
| assert actual_foo is expected_foo | ||
| def test_le(self): | ||
| json = jsonify({ | ||
| 'column': 'foo', | ||
| 'value': 15, | ||
| 'operator': '<=' | ||
| }) | ||
| actual_foos = jsonquery(self.session, self.model, json).all() | ||
| expected_foos = self.query.filter(self.model.foo <= 15).all() | ||
| assert 2 == len(actual_foos) == len(expected_foos) | ||
| assert set(actual_foos) == set(expected_foos) | ||
| def test_gt(self): | ||
| json = jsonify({ | ||
| 'column': 'foo', | ||
| 'value': 15, | ||
| 'operator': '>' | ||
| }) | ||
| actual_foo = jsonquery(self.session, self.model, json).one() | ||
| expected_foo = self.query.filter(self.model.foo > 15).one() | ||
| assert actual_foo is expected_foo | ||
| def test_ge(self): | ||
| json = jsonify({ | ||
| 'column': 'foo', | ||
| 'value': 15, | ||
| 'operator': '>=' | ||
| }) | ||
| actual_foos = jsonquery(self.session, self.model, json).all() | ||
| expected_foos = self.query.filter(self.model.foo >= 15).all() | ||
| assert 2 == len(actual_foos) == len(expected_foos) | ||
| assert set(actual_foos) == set(expected_foos) | ||
| class StringColumnTestCase(unittest.TestCase): | ||
| def setUp(self): | ||
| Base = declarative_base() | ||
| class Foo(Base): | ||
| __tablename__ = 'foos' | ||
| id = Column(Integer, primary_key=True) | ||
| foo = Column(String) | ||
| engine = create_engine("sqlite://", echo=True) | ||
| Base.metadata.create_all(engine) | ||
| self.model = Foo | ||
| self.session = sessionmaker(bind=engine)() | ||
| self.add_foo(u'Hello') | ||
| self.add_foo(u'hello') | ||
| self.add_foo('HelloWorld') | ||
| self.add_foo('helloworld') | ||
| self.add_foo('HelloWorldString') | ||
| self.add_foo('helloworldstring') | ||
| def tearDown(self): | ||
| self.session.close() | ||
| def add_foo(self, value): | ||
| foo = self.model(foo=value) | ||
| self.session.add(foo) | ||
| self.session.commit() | ||
| @property | ||
| def query(self): | ||
| return self.session.query(self.model) | ||
| def like_value(self, value): | ||
| json = jsonify({ | ||
| 'column': 'foo', | ||
| 'value': value, | ||
| 'operator': 'like' | ||
| }) | ||
| actual_foos = jsonquery(self.session, self.model, json).all() | ||
| expected_foos = self.query.filter(self.model.foo.like(value)).all() | ||
| return actual_foos, expected_foos | ||
| def test_basic_like_ignores_case(self): | ||
| ''' | ||
| Test that like, ilike for a basic sqlite String column both ignore case | ||
| Passing this test indicates that like, ilike are handled identically, | ||
| and is the reason the various wildcard tests below do not have ilike | ||
| versions. | ||
| Should this test start failing, it may be because case sensitivity becomes the | ||
| default for Column(String), in which case it would be relevant to have | ||
| new like/ilike tests again. | ||
| ''' | ||
| actual, expected = self.like_value('Hello') | ||
| assert 2 == len(actual) == len(expected) | ||
| assert set(actual) == set(expected) | ||
| def test_prefix(self): | ||
| actual, expected = self.like_value('Hello%') | ||
| assert 6 == len(actual) == len(expected) | ||
| assert set(actual) == set(expected) | ||
| def test_suffix(self): | ||
| actual, expected = self.like_value('%World') | ||
| assert 2 == len(actual) == len(expected) | ||
| assert set(actual) == set(expected) | ||
| def test_prefix_and_suffix(self): | ||
| actual, expected = self.like_value('%World%') | ||
| assert 4 == len(actual) == len(expected) | ||
| assert set(actual) == set(expected) |
Alert delta unavailable
Currently unable to show alert delta for PyPI packages.
Improved metrics
- Number of package files
12
140%Worsened metrics
- Total package byte prevSize
27991
-1.95%- Lines of code
226
-54.25%