Repository URL to install this package:
|
Version:
1:7.26.0-1 ▾
|
datadog-agent
/
opt
/
datadog-agent
/
embedded
/
lib
/
python3.8
/
site-packages
/
ijson
/
common.py
|
|---|
'''
Backend independent higher level interfaces, common exceptions.
'''
import decimal
class JSONError(Exception):
'''
Base exception for all parsing errors.
'''
pass
class IncompleteJSONError(JSONError):
'''
Raised when the parser can't read expected data from a stream.
'''
pass
def parse(basic_events):
'''
An iterator returning parsing events with the information about their location
with the JSON object tree. Events are tuples ``(prefix, type, value)``.
Available types and values are:
('null', None)
('boolean', <True or False>)
('number', <int or Decimal>)
('string', <unicode>)
('map_key', <str>)
('start_map', None)
('end_map', None)
('start_array', None)
('end_array', None)
Prefixes represent the path to the nested elements from the root of the JSON
document. For example, given this document::
{
"array": [1, 2],
"map": {
"key": "value"
}
}
the parser would yield events:
('', 'start_map', None)
('', 'map_key', 'array')
('array', 'start_array', None)
('array.item', 'number', 1)
('array.item', 'number', 2)
('array', 'end_array', None)
('', 'map_key', 'map')
('map', 'start_map', None)
('map', 'map_key', 'key')
('map.key', 'string', u'value')
('map', 'end_map', None)
('', 'end_map', None)
'''
path = []
for event, value in basic_events:
if event == 'map_key':
prefix = '.'.join(path[:-1])
path[-1] = value
elif event == 'start_map':
prefix = '.'.join(path)
path.append(None)
elif event == 'end_map':
path.pop()
prefix = '.'.join(path)
elif event == 'start_array':
prefix = '.'.join(path)
path.append('item')
elif event == 'end_array':
path.pop()
prefix = '.'.join(path)
else: # any scalar value
prefix = '.'.join(path)
yield prefix, event, value
class ObjectBuilder(object):
'''
Incrementally builds an object from JSON parser events. Events are passed
into the `event` function that accepts two parameters: event type and
value. The object being built is available at any time from the `value`
attribute.
Example::
>>> from ijson import basic_parse
>>> from ijson.common import ObjectBuilder
>>> from ijson.compat import BytesIO
>>> builder = ObjectBuilder()
>>> f = BytesIO(b'{"key": "value"}')
>>> for event, value in basic_parse(f):
... builder.event(event, value)
>>> builder.value == {'key': 'value'}
True
'''
def __init__(self, map_type=None):
def initial_set(value):
self.value = value
self.containers = [initial_set]
self.map_type = map_type or dict
def event(self, event, value):
if event == 'map_key':
self.key = value
elif event == 'start_map':
mappable = self.map_type()
self.containers[-1](mappable)
def setter(value):
mappable[self.key] = value
self.containers.append(setter)
elif event == 'start_array':
array = []
self.containers[-1](array)
self.containers.append(array.append)
elif event == 'end_array' or event == 'end_map':
self.containers.pop()
else:
self.containers[-1](value)
def items(prefixed_events, prefix, map_type=None):
'''
An iterator returning native Python objects constructed from the events
under a given prefix.
'''
prefixed_events = iter(prefixed_events)
try:
while True:
current, event, value = next(prefixed_events)
if current == prefix:
if event in ('start_map', 'start_array'):
builder = ObjectBuilder(map_type=map_type)
end_event = event.replace('start', 'end')
while (current, event) != (prefix, end_event):
builder.event(event, value)
current, event, value = next(prefixed_events)
del builder.containers[:]
yield builder.value
else:
yield value
except StopIteration:
pass
def kvitems(prefixed_events, prefix, map_type=None):
'''
An iterator returning (key, value) pairs constructed from the events
under a given prefix. The prefix should point to JSON objects
'''
prefixed_events = iter(prefixed_events)
builder = None
try:
while True:
path, event, value = next(prefixed_events)
while path == prefix and event == 'map_key':
key = value
builder = ObjectBuilder(map_type=map_type)
path, event, value = next(prefixed_events)
while path != prefix:
builder.event(event, value)
path, event, value = next(prefixed_events)
del builder.containers[:]
yield key, builder.value
except StopIteration:
pass
def number(str_value):
'''
Converts string with a numeric value into an int or a Decimal.
Used in different backends for consistent number representation.
'''
if not ('.' in str_value or 'e' in str_value or 'E' in str_value):
return int(str_value)
return decimal.Decimal(str_value)