"""
Module for basic field and field-configurable object scaffolding
"""
# pyright: basic
import datetime
import io
import typing
import warnings
from collections import defaultdict
from collections.abc import Mapping
from enum import Flag, auto
from pathlib import Path
from httpx._types import FileContent # type: ignore
from typing_extensions import override
from sf_toolkit.logger import getLogger
T = typing.TypeVar("T")
U = typing.TypeVar("U")
_LOGGER = getLogger("data")
[docs]
class ReadOnlyAssignmentException(TypeError):
"""Exception for when value assignments are performed on readonly fields"""
[docs]
class MultiPicklistValue(str):
"""type for semicolon-delimited multi-value attributes"""
values: list[str]
[docs]
def __init__(self, source: str):
self.values = (source and source.split(";")) or []
@override
def __str__(self):
return ";".join(self.values)
[docs]
class FieldFlag(Flag):
"""Flags to describe the allowed functionality on defined fields"""
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:
"""
Base object to be extended with Field definitions.
"""
_fields: typing.ClassVar[dict[str, "Field[type[typing.Any]]"]]
_strict_fields: typing.ClassVar[bool] = False
_values: dict[str, typing.Any]
_dirty_fields: set[str]
[docs]
def __init__(
self,
**field_values: typing.Any,
):
self._values = {}
_fields = object_fields(type(self))
self._dirty_fields = set()
for field, value in field_values.items():
if field not in _fields:
message = f"Field {field} not defined for {type(self).__qualname__}"
if type(self)._strict_fields:
raise KeyError(message)
else:
warnings.warn(message)
setattr(self, field, value)
for field_name, field in _fields.items():
if (
field_name not in field_values
and (_default := field.default) is not None
):
# set default values for fields not provided
if (
callable(field.default)
and field.meta_py_type
and not isinstance(field.default, field.meta_py_type)
):
# if a method/function/lambda is passed, call that to get default value
_default = field.default()
setattr(self, field_name, _default)
dirty_fields(self).clear()
def __init_subclass__(cls, strict_fields: bool = False) -> None:
cls_fields = object_fields(cls)
for parent in cls.__mro__:
if (
issubclass(parent, FieldConfigurableObject)
and parent is not FieldConfigurableObject
):
parent_fields = object_fields(parent)
for field, fieldtype in parent_fields.items():
if field not in cls_fields:
cls_fields[field] = fieldtype
cls._strict_fields = strict_fields
def __getitem__(self, name: str):
value = getattr(self, name, None)
if value is None and name not in object_fields(type(self)):
raise KeyError(f"Undefined field {name} on object {type(self)}")
return value
def __setitem__(self, name: str, value: typing.Any):
if name not in object_fields(type(self)):
raise KeyError(f"Undefined field {name} on object {type(self)}")
setattr(self, name, value)
@override
def __delattr__(self, name: str, /) -> None:
if name not in object_fields(type(self)):
raise KeyError(f"Undefined field {name} on object {type(self)}")
if name in self._values:
del self._values[name]
dirty_fields(self).add(name)
def __delitem__(self, name: str):
self.__delattr__(name)
def __str__(self):
return (
f"<{type(self).__name__} "
+ ", ".join(f"{name}={repr(value)}" for name, value in self._values.items())
+ ">"
)
_field_map: dict[type[FieldConfigurableObject], dict[str, "Field[typing.Any]"]] = (
defaultdict(dict)
)
[docs]
def object_fields(
cls: type[FieldConfigurableObject],
) -> dict[str, "Field[typing.Any]"]:
"""
returns the dictionary of field name to Field instances
configured on this class
"""
return _field_map[cls]
[docs]
def object_values(rec: FieldConfigurableObject) -> Mapping[str, typing.Any]:
"""
returns the dictionary of field name to current value
configured on this object instance
"""
return {name: getattr(rec, name, None) for name in object_fields(type(rec))}
[docs]
def query_fields(cls: type[FieldConfigurableObject]) -> list[str]:
"""
returns the list of fully qualified fields as they would
need to appear in a SOQL query
"""
fields: list[str] = list()
for field, fieldtype in object_fields(cls).items():
if (
isinstance(fieldtype, ReferenceField)
and fieldtype.meta_py_type is not None
and issubclass(fieldtype.meta_py_type, FieldConfigurableObject)
):
fields.extend(
[
field + "." + subfield
for subfield in query_fields(fieldtype.meta_py_type)
]
)
else:
fields.append(field)
return fields
[docs]
def dirty_fields(rec: FieldConfigurableObject) -> set[str]:
"""
Returns the set of fields that have been modified since the last
time this object was initialized or `save()`-d to Salesforce
"""
_dirty: set[str] | None = getattr(rec, "_dirty_fields", None)
if _dirty is None:
_dirty = set()
setattr(rec, "_dirty_fields", _dirty)
return _dirty
[docs]
def serialize_object(
record: FieldConfigurableObject,
only_changes: bool = False,
all_fields: bool = False,
include_fields: list[str] | None = None,
):
"""
Serialize this record to standard python types (dict, list, str, etc.)
for easier transport
with file formats like JSON, YAML
Args:
record: The record to serialize
only_changes: If True, only include fields that have been modified
all_fields: If True, include all fields regardless of value or flags
include_fields: A list of field names that should always be included
in the output, regardless of readonly or dirty status. Useful for
preserving reference fields like Id or external ID fields.
"""
assert not (only_changes and all_fields), (
"Cannot serialize both only changes and all fields."
)
_include = set(include_fields) if include_fields else set()
values = record._values
fields = object_fields(type(record))
dirty = dirty_fields(record)
if all_fields:
return {
name: field.format(values.get(name, None)) for name, field in fields.items()
}
if only_changes:
return {
name: field.format(values.get(name, None))
for name, field in fields.items()
if name in _include
or (name in dirty and FieldFlag.readonly not in field.flags)
}
return {
name: field.format(values.get(name, None))
for name, field in fields.items()
if name in _include
or (FieldFlag.readonly not in field.flags and (name in values or name in dirty))
}
_FCO_Type = typing.TypeVar("_FCO_Type", bound=FieldConfigurableObject)
[docs]
class FieldProps(typing.Generic[T], typing.TypedDict):
default: typing.NotRequired[T | typing.Callable[[], T] | None]
[docs]
class Field(typing.Generic[T]):
"""
Base class for all configurable field types
"""
_owner: type
_py_type: type[T] | None = None
_name: str
flags: set[FieldFlag]
default: T | typing.Callable[[], T] | None
[docs]
def __init__(
self, *flags: FieldFlag, py_type: type[T], **props: typing.Unpack[FieldProps[T]]
):
self._py_type = py_type
self.flags = set(flags)
self._owner = type(None)
self._name = ""
if (default := props.pop("default", None)) is not None:
assert isinstance(default, py_type) or callable(default), (
f"default value must be of type {py_type}"
f" or a callable generating {py_type}"
)
self.default = default
# Add descriptor protocol methods
def __get__(self, obj: FieldConfigurableObject, objtype=None) -> T:
return obj._values.get(self._name) # pyright: ignore[reportPrivateUsage, reportReturnType]
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
dirty_fields(obj).add(self._name)
def __repr__(self):
return f"<{type(self).__name__} {self._owner.__name__}.{self._name}>"
@property
def meta_py_type(self) -> type[T] | None:
"""Get the configured underlying Python type of this field's content"""
return self._py_type
[docs]
def revive(self, value: typing.Any) -> T | None:
"""
Attempts to "revive" value to be assigned to this field
into a more useful type.
"""
return value
[docs]
def __set_name__(self, cls: type[FieldConfigurableObject], name):
"""Lifecycle hook implicitly called"""
self._owner = cls
self._name = name
object_fields(cls)[name] = self
def __delete__(self, obj: FieldConfigurableObject):
del obj._values[self._name]
if hasattr(obj, "_dirty_fields"):
dirty_fields(obj).discard(self._name)
[docs]
def validate(self, value):
"""Validates the revived value passed to the field"""
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__} {str(value)[:50]}"
)
[docs]
class RawField(Field[typing.Any]):
"""
A Field that does no transformation or validation on the values passed to it.
"""
[docs]
def __init__(
self, *flags: FieldFlag, **props: typing.Unpack[FieldProps[typing.Any]]
):
super().__init__(*flags, py_type=object, **props)
[docs]
@override
def validate(self, value):
return
[docs]
class TextField(Field[str]):
"""
A field to contain text or string values
"""
[docs]
def __init__(self, *flags: FieldFlag, **props: typing.Unpack[FieldProps[str]]):
super().__init__(*flags, py_type=str, **props)
[docs]
class IdField(TextField):
"""
A field to contain 15- or 18-character alphanumeric Id
Strings used by Salesforce
"""
[docs]
@override
def validate(self, value: str | None):
if value is None:
return None
message = f" '{value}' is not a valid Salesforce Id. "
if not isinstance(value, str):
raise TypeError(message + "Expected a string.")
if len(value) not in (
15,
18,
):
raise ValueError(
message + f"Expected a string of length 15 or 18, found {len(value)}"
)
if not value.isalnum():
raise ValueError(message + "Expected strictly alphanumeric characters.")
[docs]
class PicklistField(TextField):
"""
A field to contain text values chosen from a pre-configured list.
"""
_options_: list[str]
[docs]
def __init__(
self,
*flags: FieldFlag,
options: list[str] | None = None,
**props: typing.Unpack[FieldProps[str]],
):
super().__init__(*flags, **props)
self._options_ = options or []
if (default_value := props.get("default", None)) is not None:
if (
options
and isinstance(default_value, str)
and default_value not in options
):
raise ValueError(
(
f"Default value '{default_value}' is not in configured values for field"
f" {self._name}"
)
)
[docs]
@override
def validate(self, value: str):
if self._options_ and value not in self._options_:
raise ValueError(
(
f"Selection '{value}' is not in "
f"configured values for field {self._name}"
)
)
[docs]
class MultiPicklistField(Field[MultiPicklistValue]):
"""
A field to contain text values (optionally more than one)
chosen from a pre-configured list
"""
_options_: list[str]
[docs]
def __init__(
self,
*flags: FieldFlag,
options: list[str] | None = None,
**props: typing.Unpack[FieldProps[MultiPicklistValue]],
):
super().__init__(*flags, py_type=MultiPicklistValue, **props)
if (default_value := props.get("default", None)) is not None:
if isinstance(default_value, str):
default_value = props["default"] = MultiPicklistValue(default_value)
if options and isinstance(default_value, MultiPicklistValue):
for value in default_value.values:
if value not in options:
raise ValueError(
(
f"Default value '{value}' is not in configured values for field"
f" {self._name}"
)
)
self._options_ = options or []
[docs]
@override
def revive(self, value: str):
return MultiPicklistValue(value)
[docs]
@override
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]):
"""
A field to contain a floating-point numeric value.
"""
[docs]
def __init__(self, *flags: FieldFlag, **props: typing.Unpack[FieldProps[float]]):
super().__init__(*flags, py_type=float, **props)
[docs]
@override
def revive(self, value: typing.Any):
return None if value is None else float(value)
[docs]
class IntField(Field[int]):
"""
A field to contain an integer numeric value.
"""
[docs]
def __init__(self, *flags: FieldFlag, **props: typing.Unpack[FieldProps[int]]):
super().__init__(*flags, py_type=int, **props)
[docs]
@override
def revive(self, value: typing.Any):
return None if value is None else int(value)
[docs]
class CheckboxField(Field[bool]):
"""
A field to contain a boolean value.
"""
_truthy_values = frozenset(("true", "1", "yes"))
[docs]
def __init__(self, *flags: FieldFlag, **props: typing.Unpack[FieldProps[bool]]):
super().__init__(*flags, py_type=bool, **props)
[docs]
@override
def revive(self, value: typing.Any):
if value is None:
return None
if isinstance(value, str):
return value.lower() in self._truthy_values
return bool(value)
[docs]
class DateField(Field[datetime.date]):
"""
A field to contain a date value.
"""
[docs]
def __init__(
self, *flags: FieldFlag, **props: typing.Unpack[FieldProps[datetime.date]]
):
super().__init__(*flags, py_type=datetime.date, **props)
[docs]
@override
def revive(self, value: typing.Any):
if value is None or isinstance(value, datetime.date):
return value
return datetime.date.fromisoformat(value)
[docs]
class TimeField(Field[datetime.time]):
"""
A field to contain a time value.
"""
[docs]
def __init__(
self, *flags: FieldFlag, **props: typing.Unpack[FieldProps[datetime.time]]
):
super().__init__(*flags, py_type=datetime.time, **props)
[docs]
@override
def revive(self, value: typing.Any):
if value:
return datetime.time.fromisoformat(str(value))
return None
[docs]
class DateTimeField(Field[datetime.datetime]):
"""
A field to contain a datetime value.
"""
[docs]
def __init__(
self, *flags: FieldFlag, **props: typing.Unpack[FieldProps[datetime.datetime]]
):
super().__init__(*flags, py_type=datetime.datetime, **props)
[docs]
@override
def revive(self, value: str | None):
if value is None:
return None
return datetime.datetime.fromisoformat(str(value))
[docs]
class ReferenceField(Field[_FCO_Type]):
"""
A field to contain a nested field configurable object,
typically represented in Salesforce as a lookup or master-detail relationship
"""
[docs]
@override
def __set_name__(self, cls: type[FieldConfigurableObject], name: str):
"""Lifecycle hook implicitly called"""
_folded = name.casefold()
if _folded.endswith("id") or _folded.endswith("__c"):
_LOGGER.warning(
f"Reference Field {cls.__qualname__}.{name} likely pointing at ID value. "
)
self._owner = cls
self._name = name
object_fields(cls)[name] = self
[docs]
@override
def revive(self, value: typing.Any): # pyright: ignore[reportIncompatibleMethodOverride]
if value is None:
return None
assert self._py_type is not None
if isinstance(value, self._py_type):
return value
if isinstance(value, dict):
return self._py_type(**value)
return value
[docs]
class ListField(Field[list[_FCO_Type]]):
"""
A field to contain a nested list of field configurable object,
typically represented in Salesforce as a lookup or master-detail relationship
"""
_nested_type: type[_FCO_Type | typing.Any]
[docs]
def __init__(
self,
item_type: type[_FCO_Type | typing.Any],
*flags: FieldFlag,
**props: typing.Unpack[FieldProps[list[_FCO_Type]]],
):
self._nested_type = item_type
super().__init__(*flags, py_type=list, **props)
try:
global SObjectList, SObject
# ensure SObjectList is imported
# at the time of SObject type/class definition
_ = SObjectList
except NameError:
from .sobject import SObject, SObjectList
[docs]
def revive(
self,
value: list[dict[str, typing.Any] | _FCO_Type] | dict[str, typing.Any] | None,
): # type: ignore
if value is None:
return None
if isinstance(value, SObjectList):
return value
if isinstance(value, list):
if issubclass(self._nested_type, SObject):
return SObjectList( # type: ignore
(
self._nested_type(**object_values(item))
if isinstance(item, FieldConfigurableObject)
else self._nested_type(**item)
for item in value
)
) # type: ignore
else:
return value
if isinstance(value, dict):
# assume the dict is a QueryResult-formatted dictionary
if issubclass(self._nested_type, SObject):
return SObjectList(
[self._nested_type(**item) for item in value["records"]]
) # type: ignore
return list(value.items())
raise TypeError(
f"Unexpected type {type(value)} for {type(self).__name__}[{self._nested_type.__name__}]"
)
[docs]
class BlobData:
"""Class to represent blob data that will be uploaded to Salesforce"""
_filepointer: io.IOBase | None = None
[docs]
def __init__(
self,
data: typing.Union[str, bytes, Path, io.IOBase],
filename: str | None = None,
content_type: str | None = None,
**props: typing.Unpack[FieldProps[datetime.date]],
):
self.data = data
self.filename = filename
self.content_type = content_type
# Determine filename if not provided
if self.filename is None:
if isinstance(data, Path):
self.filename = data.name
# Determine content type if not provided
if self.content_type is None:
if self.filename and "." in self.filename:
ext = self.filename.split(".")[-1].lower()
if ext in ["pdf", "doc", "docx", "xls", "xlsx", "ppt", "pptx"]:
self.content_type = f"application/{ext}"
elif ext in ["jpg", "jpeg", "png", "gif"]:
self.content_type = f"image/{ext}"
else:
self.content_type = "application/octet-stream"
else:
self.content_type = "application/octet-stream"
[docs]
def __enter__(self) -> FileContent:
"""Get the binary content of the blob data"""
if isinstance(self.data, str):
return self.data.encode("utf-8")
elif isinstance(self.data, bytes):
return self.data
elif isinstance(self.data, Path):
self._filepointer = self.data.open()
with open(self.data, "rb") as f:
return f.read()
elif isinstance(self.data, io.IOBase): # pyright: ignore[reportUnnecessaryIsInstance]
# Reset the file pointer if it's a file object
if hasattr(self.data, "seek"):
self.data.seek(0)
return self.data.read()
else:
raise TypeError(f"Unsupported data type: {type(self.data)}")
def __exit__(self, exc_type, exc_value, traceback):
if self._filepointer:
self._filepointer.close()
[docs]
class GeolocationSerialized(typing.TypedDict):
latitude: float
longitude: float
[docs]
class Geolocation(typing.NamedTuple):
latitude: float
longitude: float
[docs]
class GeolocationField(Field[Geolocation]):
"""Field type for handling geolocation data in Salesforce"""
[docs]
def __init__(
self, *flags: FieldFlag, **props: typing.Unpack[FieldProps[Geolocation]]
):
super().__init__(*flags, py_type=Geolocation, **props)
[docs]
@override
def revive(self, value: Geolocation | GeolocationSerialized | None):
if value is None:
return None
if isinstance(value, Geolocation):
return value
if isinstance(value, dict):
return Geolocation(
latitude=value.get("latitude"), longitude=value.get("longitude")
)
raise TypeError(f"Cannot revive value of type {type(value)} to Geolocation")
[docs]
class AddressSerialized(typing.TypedDict):
Accuracy: typing.NotRequired[str]
City: typing.NotRequired[str]
Country: typing.NotRequired[str]
CountryCode: typing.NotRequired[str]
Latitude: typing.NotRequired[float]
Longitude: typing.NotRequired[float]
PostalCode: typing.NotRequired[str]
State: typing.NotRequired[str]
StateCode: typing.NotRequired[str]
Street: typing.NotRequired[str]
[docs]
class Address(typing.NamedTuple):
Accuracy: str | None = None
City: str | None = None
Country: str | None = None
CountryCode: str | None = None
Latitude: float | None = None
Longitude: float | None = None
PostalCode: str | None = None
State: str | None = None
StateCode: str | None = None
Street: str | None = None
[docs]
class AddressField(Field[Address]):
"""
Field type for handling address data in Salesforce
"""
[docs]
def __init__(self, *flags: FieldFlag, **props: typing.Unpack[FieldProps[Address]]):
super().__init__(*flags, py_type=Address, **props)
[docs]
@override
def revive(self, value: Address | AddressSerialized | None):
if value is None:
return None
if isinstance(value, Address):
return value
if isinstance(value, dict):
return Address(
Accuracy=value.get("Accuracy"),
City=value.get("City"),
Country=value.get("Country"),
CountryCode=value.get("CountryCode"),
Latitude=value.get("Latitude"),
Longitude=value.get("Longitude"),
PostalCode=value.get("PostalCode"),
State=value.get("State"),
StateCode=value.get("StateCode"),
Street=value.get("Street"),
)
raise TypeError(f"Cannot revive value of type {type(value)} to Address")
[docs]
class BlobField(Field[BlobData]):
"""Field type for handling blob data in Salesforce"""
[docs]
def __init__(self, *flags: FieldFlag, **props: typing.Unpack[FieldProps[BlobData]]):
super().__init__(*flags, py_type=BlobData, **props)
[docs]
def revive(self, value):
if value is None:
return None
if isinstance(value, BlobData):
return value
# Convert different input types to BlobData
return BlobData(value)
# Add descriptor protocol methods
def __get__(self, obj: FieldConfigurableObject, objtype=None) -> BlobData:
if obj is None:
return self
return getattr(obj, self._name + "_BlobData", None) # 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__}"
)
setattr(obj, self._name + "_BlobData", value)
dirty_fields(obj).add(self._name)
FIELD_TYPE_LOOKUP: dict[str, type[Field[typing.Any]]] = {
"boolean": CheckboxField,
"id": IdField,
"string": TextField,
"phone": TextField,
"url": TextField,
"email": TextField,
"textarea": TextField,
"picklist": PicklistField,
"multipicklist": MultiPicklistField,
"reference": ReferenceField,
"currency": NumberField,
"double": NumberField,
"percent": NumberField,
"int": IntField,
"date": DateField,
"datetime": DateTimeField,
"time": TimeField,
"blob": BlobField,
"base64": BlobField,
"location": GeolocationField,
}