Repository URL to install this package:
|
Version:
3.0.0rc10 ▾
|
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