import datetime
from enum import Flag, auto
import typing
T = typing.TypeVar("T")
U = typing.TypeVar("U")
[docs]
class ReadOnlyAssignmentException(TypeError): ...
[docs]
class SObjectFieldDescribe(typing.NamedTuple):
"""Represents metadata about a Salesforce SObject field"""
name: str
label: str
type: str
length: int = 0
nillable: bool = False
picklistValues: list[dict] = []
referenceTo: list[str] = []
relationshipName: str | None = None
unique: bool = False
updateable: bool = False
createable: bool = False
defaultValue: typing.Any = None
externalId: bool = False
autoNumber: bool = False
calculated: bool = False
caseSensitive: bool = False
dependentPicklist: bool = False
deprecatedAndHidden: bool = False
displayLocationInDecimal: bool = False
filterable: bool = False
groupable: bool = False
permissionable: bool = False
restrictedPicklist: bool = False
sortable: bool = False
writeRequiresMasterRead: bool = False
[docs]
class MultiPicklistValue(str):
values: list[str]
[docs]
def __init__(self, source: str):
self.values = source.split(";")
def __str__(self):
return ";".join(self.values)
[docs]
class FieldFlag(Flag):
nillable = auto()
unique = auto()
readonly = auto()
case_sensitive = auto()
updateable = auto()
createable = auto()
calculated = auto()
filterable = auto()
sortable = auto()
groupable = auto()
permissionable = auto()
restricted_picklist = auto()
display_location_in_decimal = auto()
write_requires_master_read = auto()
[docs]
class FieldConfigurableObject:
_values: dict[str, typing.Any]
_dirty_fields: set[str]
_fields: typing.ClassVar[dict[str, "Field"]]
[docs]
def __init__(self):
self._values = {}
self._dirty_fields = set()
[docs]
@classmethod
def keys(cls) -> typing.Iterable[str]:
assert hasattr(cls, "_fields"), f"No Field definitions found for class {cls.__name__}"
return cls._fields.keys()
[docs]
@classmethod
def query_fields(cls) -> list[str]:
assert hasattr(cls, "_fields"), f"No Field definitions found for class {cls.__name__}"
fields = list()
for field, fieldtype in cls._fields.items():
if isinstance(fieldtype, ReferenceField) and fieldtype._py_type:
fields.extend(
[
field + "." + subfield
for subfield in fieldtype._py_type.query_fields()
]
)
else:
fields.append(field)
return fields
@property
def dirty_fields(self) -> set[str]:
return self._dirty_fields
@dirty_fields.deleter
def dirty_fields(self):
self._dirty_fields = set()
[docs]
def serialize(self, only_changes: bool = False, all_fields: bool = False):
assert not (only_changes and all_fields), (
"Cannot serialize both only changes and all fields."
)
if all_fields:
return {
name: field.format(self._values.get(name, None))
for name, field in self._fields.items()
}
if only_changes:
return {
name: field.format(value)
for name, value in self._values.items()
if (field := self._fields[name])
and name in self.dirty_fields
and FieldFlag.readonly not in field.flags
}
return {
name: field.format(value)
for name, value in self._values.items()
if (field := self._fields[name]) and FieldFlag.readonly not in field.flags
}
def __getitem__(self, name):
if name not in self.keys():
raise KeyError(f"Undefined field {name} on object {type(self)}")
return getattr(self, name, None)
def __setitem__(self, name, value):
if name not in self.keys():
raise KeyError(f"Undefined field {name} on object {type(self)}")
setattr(self, name, value)
[docs]
class Field(typing.Generic[T]):
_py_type: type[T] | None = None
flags: set[FieldFlag]
[docs]
def __init__(self, py_type: type[T], *flags: FieldFlag):
self._py_type = py_type
self.flags = set(flags)
# Add descriptor protocol methods
def __get__(self, obj: FieldConfigurableObject, objtype=None) -> T:
if obj is None:
return self
return obj._values.get(self._name) # type: ignore
def __set__(self, obj: FieldConfigurableObject, value: typing.Any):
value = self.revive(value)
self.validate(value)
if FieldFlag.readonly in self.flags and self._name in obj._values:
raise ReadOnlyAssignmentException(
f"Field {self._name} is readonly on object {self._owner.__name__}"
)
obj._values[self._name] = value
obj.dirty_fields.add(self._name)
[docs]
def revive(self, value: typing.Any):
return value
def __set_name__(self, cls: type[FieldConfigurableObject], name):
self._owner = cls
self._name = name
if not hasattr(cls, "_fields"):
cls._fields = {}
cls._fields[name] = self
def __delete__(self, obj: FieldConfigurableObject):
del obj._values[self._name]
if hasattr(obj, "_dirty_fields"):
obj._dirty_fields.discard(self._name)
[docs]
def validate(self, value):
if value is None:
return
if self._py_type is not None and not isinstance(value, self._py_type):
raise TypeError(
f"Expected {self._py_type.__qualname__} for field {self._name} "
f"on {self._owner.__name__}, got {type(value).__name__}"
)
[docs]
class TextField(Field[str]):
[docs]
def __init__(self, *flags: FieldFlag):
super().__init__(str, *flags)
[docs]
class IdField(TextField):
[docs]
def validate(self, value):
if value is None:
return
assert isinstance(value, str), (
f" '{value}' is not a valid Salesforce Id. Expected a string."
)
assert len(value) in (15, 18), (
f" '{value}' is not a valid Salesforce Id. Expected a string of length 15 or 18, found {len(value)}"
)
assert value.isalnum(), (
f" '{value}' is not a valid Salesforce Id. Expected strictly alphanumeric characters."
)
[docs]
class PicklistField(TextField):
_options_: list[str]
[docs]
def __init__(self, *flags: FieldFlag, options: list[str] | None = None):
super().__init__(*flags)
self._options_ = options or []
[docs]
def validate(self, value: str):
if self._options_ and value not in self._options_:
raise ValueError(
f"Selection '{value}' is not in configured values for field {self._name}"
)
[docs]
class MultiPicklistField(Field[MultiPicklistValue]):
_options_: list[str]
[docs]
def __init__(self, *flags: FieldFlag, options: list[str] | None = None):
super().__init__(MultiPicklistValue, *flags)
self._options_ = options or []
[docs]
def revive(self, value: str):
return MultiPicklistValue(value)
[docs]
def validate(self, value: MultiPicklistValue):
for item in value.values:
if self._options_ and item not in self._options_:
raise ValueError(
f"Selection '{item}' is not in configured values for {self._name}"
)
[docs]
class NumberField(Field[float]):
[docs]
def __init__(self, *flags: FieldFlag):
super().__init__(float, *flags)
[docs]
class IntField(Field[int]):
[docs]
def __init__(self, *flags: FieldFlag):
super().__init__(int, *flags)
[docs]
class CheckboxField(Field[bool]):
[docs]
def __init__(self, *flags: FieldFlag):
super().__init__(bool, *flags)
[docs]
class DateField(Field[datetime.date]):
[docs]
def __init__(self, *flags: FieldFlag):
super().__init__(datetime.date, *flags)
[docs]
def revive(self, value: datetime.date | str):
if isinstance(value, datetime.date):
return value
return datetime.date.fromisoformat(value)
[docs]
class TimeField(Field[datetime.time]):
[docs]
def __init__(self, *flags: FieldFlag):
super().__init__(datetime.time, *flags)
[docs]
class DateTimeField(Field[datetime.datetime]):
[docs]
def __init__(self, *flags: FieldFlag):
super().__init__(datetime.datetime, *flags)
[docs]
def revive(self, value: str):
return datetime.datetime.fromisoformat(str(value))
[docs]
class ReferenceField(Field[T]):
[docs]
def revive(self, value):
if value is None:
return value
assert self._py_type is not None
if isinstance(value, self._py_type):
return value
if isinstance(value, dict):
return self._py_type(**value)
[docs]
class ListField(Field[list[T]]):
_nested_type: type[T]
[docs]
def __init__(self, item_type: type[T], *flags: FieldFlag):
self._nested_type = item_type
super().__init__(list, *flags)
try:
global SObjectList
# ensure SObjectList is imported at the time of SObject type/class definition
SObjectList # type: ignore
except NameError:
from .sobject import SObjectList
[docs]
def revive(self, value: list[dict | FieldConfigurableObject]):
if value is None:
return value
if isinstance(value, SObjectList):
return value
if isinstance(value, list):
return SObjectList([self._nested_type(**item) for item in value]) # type: ignore
if isinstance(value, dict):
# assume the dict is a QueryResult-formatted dictionary
return SObjectList([self._nested_type(**item) for item in value["records"]]) # type: ignore
raise TypeError(f"Unexpected type {type(value)} for {type(self).__name__}[{self._nested_type.__name__}]")
FIELD_TYPE_LOOKUP: dict[str, type[Field]] = {
"boolean": CheckboxField,
"id": IdField,
"string": TextField,
"phone": TextField,
"url": TextField,
"email": TextField,
"textarea": TextField,
"picklist": TextField,
"multipicklist": MultiPicklistField,
"reference": ReferenceField,
"currency": NumberField,
"double": NumberField,
"percent": NumberField,
"int": NumberField,
"date": DateField,
"datetime": DateTimeField,
"time": TimeField,
}