Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
tdw-catalog / tdw_catalog / entity.py
Size: Mime:
import datetime
import importlib
from inspect import signature
from typing import Any, Callable, Optional, Union, Type

import dateutil.parser

from tdw_catalog.utils import _type_is_optional, _parse_timestamp


class Property:

    def __init__(
        self,
        key: str,
        typ: Union[Type, datetime.datetime],
        relation: Union['EntityBase', str, None] = None,
        relation_key: Optional[str] = None,
        readable: bool = True,
        writable: bool = False,
        default: Optional[any] = None,
        serialize: Optional[bool] = None,
        deserialize: Union[bool, Callable] = True,
        display_key: Optional[str] = None,
    ):
        """
        Property

        Parameters
        ----------
        key : str
            The key of the property as defined in the protobuf
        display_key : Optional[str]
            An optional override "display name" for the property,
            if the name in the SDK should be different than the underlying protobuf.
            Must be a valid python class member name
        typ : type
            The type the property is for property access
        relation : Union[EntityBase, str, None]
            The class that refers to a foreign key column.  If using the string
            variant, use a full import path to the module, including the class
            name.
        relation_key: Optional[str]
            Optionally override the key that will be used to store the fetched
            relation object
        readable : bool
            Marks whether this property can be read from. False would denote
            that this property is kept internally but not accessible to users.
        writable : bool
            Marks whether this property can be written to.  False would denote
            it has readonly
        default : any
            The default value of the property if the value is None
        serialize : bool
            Marks whether the property should be serialized when updating or
            creating with the API.
        deserialize : bool | Callable
            Marks whether the property should be deserialized from the API response.
            Optionally, a function we can call to deserialize the value
        """
        self.key = key
        self.display_key = display_key
        self.type = typ
        self.relation = relation
        self.relation_key = relation_key
        self.readable = readable
        self.writable = writable
        self.default = default
        self.deserialize = deserialize

        if relation_key is not None and display_key is not None:
            raise Exception(
                'Cannot use relation_key and display_key at the same time')

        if (writable and serialize is None) or serialize:
            self.serialize = True
        else:
            self.serialize = False

    def __str__(self) -> str:
        return f"<Property {self.key}:{self.type} relation={self.relation} writable={self.writable}>"

    def relation_class(self) -> Optional['EntityBase']:
        if self.relation is None:
            return None
        elif isinstance(self.relation, str):
            module_path = self.relation.split(".")
            module = importlib.import_module(".".join(module_path[:-1]))
            self.relation = getattr(module, module_path[-1])
            return self.relation
        else:
            return self.relation

    def default_value(self) -> any:
        if self.default is None:
            # We need special treatment for datetime.  We will also protect against the case of the user
            # inadvertently specifying the module datetime instead of the type datetime.datetime
            if self.type == datetime.datetime or self.type == datetime:
                return datetime.datetime.fromtimestamp(0)
            elif callable(self.type):
                try:
                    # produce a zero value for the type
                    return self.type()
                except:
                    # If the type cannot produce a zero value, we use None
                    return None
            else:
                return None
        else:
            return self.default

    def attr(self) -> str:
        return "_" + self.key

    def property_name(self) -> str:
        return self.display_key if self.display_key is not None else self.key

    def relation_name(self) -> str:
        if self.relation is not None:
            name = ''
            # relation_key takes precedence over relation_key
            if self.relation_key is not None:
                return self.relation_key
            # use the display name instead of the key if present
            if self.display_key is not None:
                name = self.display_key.replace('_id', '')
            else:
                name = self.key.replace('_id', '')
            # if not readable, then the relation_name should be prefixed
            # with a _ for internal SDK use only
            if self.readable is False:
                name = '_' + name
            return name

    def relation_attr(self) -> str:
        if self.relation is not None:
            name = "_" + self.relation_name()
            # if not readable, the _ prefix is already taken by relation_name
            # so we double-prefix with another _
            if self.readable is False:
                name = '_' + name
            return name

    def relation_getter_attr(self) -> str:
        if self.relation is not None:
            return f"_get_{self.relation_name()}"


class Entity:

    def __init__(self, properties):
        self._properties = properties

    def __call__(self, cls):
        if hasattr(cls, "_properties"):
            cls._properties = cls._properties + (self._properties or [])
        else:
            cls._properties = self._properties or []

        self._validate_properties(cls)

        # Create the property getters and setters
        for prop in cls._properties:
            setattr(
                cls, prop.property_name(),
                property(fget=self.prop_getter_gen(prop),
                         fset=self.prop_setter_gen(prop),
                         fdel=self.prop_deleter_gen(prop)))

            # Create the relation getters and setters
            if prop.relation is not None:
                setattr(
                    cls, prop.relation_name(),
                    property(fget=self.rel_getter_gen(prop),
                             fset=self.rel_setter_gen(prop),
                             fdel=self.rel_deleter_gen(prop)))

        return cls

    def _validate_properties(self, cls):
        keys = {prop.key for prop in self._properties}
        if len(keys) != len(self._properties):
            raise SyntaxError(
                f"Duplicate property detected for entity {cls.__name__}")

        for key in keys:
            if key == "":
                raise SyntaxError(
                    f"Blank property detected for entity {cls.__name__}")

    @staticmethod
    def prop_getter_gen(prop):
        if not prop.readable:
            return None

        def f(self):
            # Get the property value
            return getattr(self, prop.attr())

        return f

    @staticmethod
    def prop_setter_gen(prop):
        if not prop.writable:
            return None

        def f(self, value):
            # Set the underlying attribute to this value
            setattr(self, prop.attr(), value)

            # If this is a relational attribute then we want to clear out the
            # relation because it will now be out of date and needs to be
            # fetched again
            if prop.relation is not None:
                setattr(self, prop.relation_attr(), None)

        return f

    @staticmethod
    def prop_deleter_gen(prop):

        def f(self):
            raise AttributeError(f"Property \"{prop.key}\" cannot be deleted")

        return f

    @staticmethod
    def rel_getter_gen(prop):

        def f(self):
            rel_id = getattr(self, prop.attr())
            if rel_id is None or rel_id == "":
                return None

            if getattr(self, prop.relation_attr()) is None:
                if hasattr(self.__class__, prop.relation_getter_attr()):
                    obj = getattr(self, prop.relation_getter_attr())()
                else:
                    obj = prop.relation_class()(self._client, id=rel_id)
                setattr(self, prop.relation_attr(), obj)

            return getattr(self, prop.relation_attr())

        return f

    @staticmethod
    def rel_setter_gen(prop):
        if not prop.writable:
            return None

        def f(self, value):
            # If we set the relation, we want to set the relation and the _id
            setattr(self, prop.attr(), value.id)
            setattr(self, prop.relation_attr(), value)

        return f

    @staticmethod
    def rel_deleter_gen(prop):

        def f(self):
            raise AttributeError(
                f"Relation \"{prop.relation_name()}\" cannot be deleted")

        return f


class EntityBase:
    _id: str
    _properties: list

    def __init__(self, client, **kwargs):
        self._client = client
        self.deserialize(kwargs)

    def __str__(self) -> str:
        return f'<{self.__class__.__name__} id={self._id}>'

    def __repr__(self) -> str:
        return self.__str__()

    def __hash__(self) -> int:
        return self.__str__().__hash__()

    def __eq__(self, other) -> bool:
        return isinstance(other, self.__class__) and self._id == other._id

    def __getitem__(self, item):
        return getattr(self, item)

    def deserialize(self, data):
        for prop in self._properties:
            if prop.deserialize is False:
                continue

            # Fetch the value off the data object and set to the relevant
            # property
            value = data.get(prop.key, prop.default_value())
            # properly parse dict-formatted dates if we receive one
            if prop.type == datetime.datetime or (
                    prop.type == Optional[datetime.datetime]
                    and value is not None):
                value = _parse_timestamp(value)
            # properlly parse custom deserializeable types if we receive one
            if callable(prop.deserialize):
                # custom deserializers can optionally take self as a secondary
                # parameter - call the two param version if it's provided,
                # otherwise call the one param version
                sig = signature(prop.deserialize)
                value = prop.deserialize(value) if len(
                    sig.parameters) == 1 else prop.deserialize(value, self)
            setattr(self, prop.attr(), value)

            # If we are deserializing a relation (takes precedence over
            # deserializing the _id field) then set the relation to this value
            # a well as the corresponding _id field
            if prop.relation is not None:
                if prop.relation_name() in data:
                    obj = data.pop(prop.relation_name())
                    setattr(self, prop.relation_attr(), obj)
                    setattr(self, prop.attr(), obj['id'])
                else:
                    setattr(self, prop.relation_attr(), None)

    def serialize(self) -> dict:
        res = {}
        for prop in self._properties:
            if prop.serialize is not True:
                continue
            value = getattr(self, prop.attr())
            if prop.type == Optional[datetime.datetime]:
                value = {
                    'is_null': True
                } if value == None else {
                    'timestamp': value.isoformat() + "Z"
                }
            if prop.type == datetime.datetime:
                value = value.isoformat() + "Z"
            if _type_is_optional(prop.type) and value is None:
                # Don't bother setting the key in the dict
                # if the value is None
                continue
            # if this is a custom type with a serialize method, then use it
            if hasattr(value, "serialize") and callable(value.serialize):
                value = value.serialize()
            # if this is a list of custom types with serialize methods, use them
            if isinstance(value, list) and len(value) > 0:
                # serialize each value in the list if it has a serialize method,
                # otherwise leave it untouched
                value = [
                    v.serialize()
                    if hasattr(v, "serialize") and callable(v.serialize) else v
                    for v in value
                ]
            res[prop.key] = value
        return res