How to use the aiozipkin.helpers.filter_none function in aiozipkin

To help you get started, we’ve selected a few aiozipkin examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github aio-libs / aiozipkin / tests / test_helpers.py View on Github external
def test_filter_none():
    r = filter_none({'a': 1, 'b': None})
    assert r == {'a': 1}

    r = filter_none({'a': 1, 'b': None, 'c': None}, keys=['a', 'c'])
    assert r == {'a': 1, 'b': None}

    r = filter_none({}, keys=['a', 'c'])
    assert r == {}

    r = filter_none({'a': 1, 'b': None, 'c': {'c': None}}, keys=['a', 'c'])
    assert r == {'a': 1, 'b': None, 'c': {'c': None}}
github aio-libs / aiozipkin / tests / test_helpers.py View on Github external
def test_filter_none():
    r = filter_none({'a': 1, 'b': None})
    assert r == {'a': 1}

    r = filter_none({'a': 1, 'b': None, 'c': None}, keys=['a', 'c'])
    assert r == {'a': 1, 'b': None}

    r = filter_none({}, keys=['a', 'c'])
    assert r == {}

    r = filter_none({'a': 1, 'b': None, 'c': {'c': None}}, keys=['a', 'c'])
    assert r == {'a': 1, 'b': None, 'c': {'c': None}}
github aio-libs / aiozipkin / aiozipkin / record.py View on Github external
rec = {
            'traceId': c.trace_id,
            'name': self._name,
            'parentId': c.parent_id,
            'id': c.span_id,
            'kind': self._kind,
            'timestamp': self._timestamp,
            'duration': self._duration,
            'debug': c.debug,
            'shared': c.shared,
            'localEndpoint': self._local_endpoint,
            'remoteEndpoint': self._remote_endpoint,
            'annotations': [a._asdict() for a in self._annotations],
            'tags': self._tags,
        }
        return filter_none(rec, ['kind'])
github aio-libs / aiozipkin / aiozipkin / record.py View on Github external
def _endpoint_asdict(endpoint: Endpoint) -> Dict[str, Any]:
    return filter_none(endpoint._asdict())