Skip to content
This repository was archived by the owner on Apr 19, 2022. It is now read-only.
This repository was archived by the owner on Apr 19, 2022. It is now read-only.

[Enhancement] Useful BaseModel class (not an issue) #9

@cmin764

Description

@cmin764

I think most "legacy" GAE users will find this useful:

"""Base models and utilities used by the derived ones."""


import datetime
import logging

import ndb_orm as ndb
from google.cloud import datastore

from <project> import app, settings


# Datastore default client settings.
PROJECT = settings.PROJECT_ID
NAMESPACE = app.config["DATASTORE_NAMESPACE"]
NDB_KWARGS = {"project": PROJECT, "namespace": NAMESPACE}

# Module level singleton client used in all DB interactions. This is lazy inited when
# is used only, so we don't have any issues with Datastore agnostic tests/debugging,
# because creating a client will require valid credentials.
client = None


class BaseModel(ndb.Model):

    """Common model properties and functionality."""

    # String used for properties with no available data.
    NOT_SET = "N/A"

    created_at = ndb.DateTimeProperty(auto_now_add=True)

    def __init__(self, *args, **kwargs):
        self._get_client()  # Activates all NDB ORM required features.
        kwargs.update(NDB_KWARGS)
        super().__init__(*args, **kwargs)

    @classmethod
    def model_name(cls):
        return cls.__name__.replace("Model", "")

    @classmethod
    def normalize(cls, value):
        if value is None:
            return cls.NOT_SET
        return value

    @staticmethod
    def _get_client():
        global client
        if not client:
            client = datastore.Client(**NDB_KWARGS)
            ndb.enable_use_with_gcd(client=client, **NDB_KWARGS)
        return client

    @classmethod
    def query(cls, **kwargs):
        query = cls._get_client().query(kind=cls._get_kind(), **kwargs)
        return query

    @classmethod
    def all(cls, query=None, keys_only=False, **kwargs):
        query = query or cls.query()
        query.order = ["-created_at"]
        if keys_only:
            query.keys_only()
        return list(query.fetch(**kwargs))

    @property
    def myself(self):
        """Return the current DB version of the same object."""
        return self.key.get()

    @property
    def exists(self):
        """Checks if the entity is saved into the Datastore."""
        try:
            return bool(self.myself) if self.key and self.key.id else False
        except Exception:
            return False

    def put(self):
        """Saving the entity into the Datastore."""
        self._get_client().put(self)
        return self.key

    @classmethod
    def put_multi(cls, entities):
        """Multiple save in the DB without interfering with `cls.put` function."""
        cls._get_client().put_multi(entities)
        return [entity.key for entity in entities]

    def remove(self):
        """Removes current entity and its dependencies (if any)."""
        self.key.delete()

    @classmethod
    def remove_multi(cls, keys):
        cls._get_client().delete_multi(keys)

    @property
    def urlsafe(self):
        return self.key.to_legacy_urlsafe().decode(settings.ENCODING)

    @classmethod
    def get(cls, urlsafe_or_key):
        if isinstance(urlsafe_or_key, (str, bytes)):
            key = ndb.Key(cls, **NDB_KWARGS)
            complete_key = key.from_legacy_urlsafe(urlsafe_or_key)
        else:
            complete_key = urlsafe_or_key

        item = complete_key.get()
        if not item:
            raise Exception("item doesn't exist")
        return item

Don't be sceptical, this is tested by these:

Just replace <project> with your package name (under a Flask app for example).

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions