Building a Python Data Access Layer

1
class Document:
2
    def __init__(self, doc, *, strict=False):
3
        self._doc = doc
4
        self._strict = strict

1
>>> relaxed_doc = Document({ 'name': 'Jones' })
2
>>> relaxed_doc.name
3
"Jones"
4
5
>>> strict_doc = Document({ 'name': 'Jones' }, strict=True)
6
>>> strict_doc.name
7
Traceback (most recent call last):
8
  File "<stdin>", line 1, in <module>
9
  File ".../docbridge/__init__.py", line 33, in __getattr__
10
    raise AttributeError(
11
AttributeError: 'Document' object has no attribute 'name'

1
def __getattr__(self, attr):
2
    if not self._strict:
3
        return self._doc[attr]
4
    else:
5
        raise AttributeError(
6
            f"{self.__class__.__name__!r} object has no attribute {attr!r}"
7
        )

1
{
2
    "cocktailName": "Old Fashioned"
3
}

1
class Cocktail(Document):
2
    cocktail_name = Field(field_name="cocktailName")

1
class Field:
2
    def __init__(self, field_name=None):
3
        """
4
        Initialize a Field attribute, mapping to an underlying BSON field.
5
6
        field_name is the name of the underlying BSON field.
7
        If field_name is None (the default), use the attribute name for lookup in the doc.
8
        """
9
        self.field_name = None
10
11
    def __set_name__(self, owner, name):
12
        """
13
        Called by Python when this Field instance is attached to a class (the owner).
14
        """
15
        self.name = name  # this is the *attribute* name on the class.
16
17
        # If no field_name was provided, then default to using the attribute
18
        # name to look up the BSON field:
19
        if self.field_name is None:
20
            self.field_name = name
21
22
    def __get__(self, ob, cls):
23
        """
24
        Called by Python when this attribute is looked up on an instance of
25
        the class it's attached to.
26
        """
27
        try:
28
            # Look up the BSON field and return it:
29
            return ob._doc[self.field_name]
30
        except KeyError as ke:
31
            raise ValueError(
32
                f"Attribute {self.name!r} is mapped to missing document property {self.field_name!r}."
33
            ) from ke

1
{
2
	"name": "Mark Smith",
3
"schema_version": 1,
4
}

1
{
2
     	"full_name": "Mark Smith"
3
	"first_name": "Mark",
4
	"last_name": "Smith",
5
	"schema_version": 2,
6
}

1
class Fallthrough:
2
    def __init__(self, field_names: Sequence[str]) -> None:
3
        self.field_names = field_names
4
5
    def __get__(self, ob, cls):
6
        for field_name in self.field_names:  # loop through the field names until one returns a value.
7
            try:
8
                return ob._doc[field_name]
9
            except KeyError:
10
                pass
11
        else:
12
            raise ValueError(
13
                f"Attribute {self.name!r} references the field names {', '.join([repr(fn) for fn in self.field_names])} which are not present."
14
            )
15
16
    def __set_name__(self, owner, name):
17
        self.name = name

1
import os
2
from docbridge import Document, Field, FallthroughField
3
from pymongo import MongoClient
4
5
collection = (
6
    MongoClient(os.environ["MDB_URI"])
7
    .get_database("docbridge_test")
8
    .get_collection("people")
9
)
10
11
collection.delete_many({})  # Clean up any leftover documents.
12
# Insert a couple of sample documents:
13
collection.insert_many(
14
    [
15
        {
16
            "name": "Mark Smith",
17
            "schema_version": 1,
18
        },
19
        {
20
            "full_name": "Mark Smith",
21
            "first_name": "Mark",
22
            "last_name": "Smith",
23
            "schema_version": 2,
24
        },
25
    ]
26
)
27
28
# Define a mapping for "person" documents:
29
class Person(Document):
30
    version = Field("schema_version")
31
    name = FallthroughField(
32
        [
33
            "name",  # v1
34
            "full_name",  # v2
35
        ]
36
    )
37
38
# This finds all the documents in the collection, but wraps each BSON document with a Person wrapper:
39
people = (Person(doc, None) for doc in collection.find())
40
for person in people:
41
    print(
42
        "Name:",
43
        person.name,
44
    )  # The name (or full_name) of the underlying document.
45
    print(
46
        "Document version:",
47
        person.version,  # The schema_version field of the underlying document.
48
    )

1
$ python examples/why/simple_example.py
2
Name: Mark Smith
3
Document version: 1
4
Name: Mark Smith
5
Document version: 2