File: dump.py

package info (click to toggle)
python-beanie 2.0.1-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 1,496 kB
  • sloc: python: 14,596; makefile: 6; sh: 6
file content (53 lines) | stat: -rw-r--r-- 1,327 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
from typing import TYPE_CHECKING, Optional, Set

from beanie.odm.utils.encoder import Encoder

if TYPE_CHECKING:
    from beanie.odm.documents import Document


def get_dict(
    document: "Document",
    to_db: bool = False,
    exclude: Optional[Set[str]] = None,
    keep_nulls: bool = True,
):
    if exclude is None:
        exclude = set()
    if document.id is None:
        exclude.add("_id")
    include = set()
    if document.get_settings().use_revision:
        include.add("revision_id")
    encoder = Encoder(
        exclude=exclude, include=include, to_db=to_db, keep_nulls=keep_nulls
    )
    return encoder.encode(document)


def get_nulls(
    document: "Document",
    exclude: Optional[Set[str]] = None,
):
    dictionary = get_dict(document, exclude=exclude, keep_nulls=True)
    return filter_none(dictionary)


def get_top_level_nones(
    document: "Document",
    exclude: Optional[Set[str]] = None,
):
    dictionary = get_dict(document, exclude=exclude, keep_nulls=True)
    return {k: v for k, v in dictionary.items() if v is None}


def filter_none(d):
    result = {}
    for k, v in d.items():
        if isinstance(v, dict):
            filtered = filter_none(v)
            if filtered:
                result[k] = filtered
        elif v is None:
            result[k] = v
    return result