import dataclasses
import datetime
import warnings
from dataclasses import dataclass
from functools import cached_property
from typing import (
TYPE_CHECKING,
Any,
ClassVar,
Dict,
Iterable,
List,
Mapping,
Optional,
Set,
Type,
Union,
cast,
)
from typing_extensions import Self as SelfType
from pyairtable.api import retrying
from pyairtable.api.api import Api, TimeoutTuple
from pyairtable.api.base import Base
from pyairtable.api.table import Table
from pyairtable.api.types import (
FieldName,
RecordDict,
RecordId,
UpdateRecordDict,
WritableFields,
)
from pyairtable.formulas import EQ, OR, RECORD_ID
from pyairtable.models import Comment
from pyairtable.orm.fields import AnyField, Field
from pyairtable.utils import datetime_from_iso_str, datetime_to_iso_str
if TYPE_CHECKING:
from builtins import _ClassInfo
[docs]class Model:
"""
Supports creating ORM-style classes representing Airtable tables.
For more details, see :ref:`orm`.
A nested class or dict called ``Meta`` is required and can specify
the following attributes:
* ``api_key`` (required) - API key or personal access token.
* ``base_id`` (required) - Base ID (not name).
* ``table_name`` (required) - Table ID or name.
* ``timeout`` - A tuple indicating a connect and read timeout. Defaults to no timeout.
* ``typecast`` - |kwarg_typecast| Defaults to ``True``.
* ``retry`` - An instance of `urllib3.util.Retry <https://urllib3.readthedocs.io/en/stable/reference/urllib3.util.html#urllib3.util.Retry>`_.
If ``None`` or ``False``, requests will not be retried.
If ``True``, the default strategy will be applied
(see :func:`~pyairtable.retry_strategy` for details).
* ``use_field_ids`` - Whether fields will be defined by ID, rather than name. Defaults to ``False``.
* ``memoize`` - Whether the model should reuse models it creates between requests.
See :ref:`Memoizing linked records` for more information.
For example, the following two are equivalent:
.. code-block:: python
from pyairtable.orm import Model, fields
class Contact(Model):
class Meta:
base_id = "appaPqizdsNHDvlEm"
table_name = "Contact"
api_key = "keyapikey"
timeout = (5, 5)
typecast = True
first_name = fields.TextField("First Name")
age = fields.IntegerField("Age")
.. code-block:: python
from pyairtable.orm import Model, fields
class Contact(Model):
Meta = {
"base_id": "appaPqizdsNHDvlEm",
"table_name": "Contact",
"api_key": "keyapikey",
"timeout": (5, 5),
"typecast": True,
}
first_name = fields.TextField("First Name")
age = fields.IntegerField("Age")
You can implement meta attributes as callables if certain values
need to be dynamically provided or are unavailable at import time:
.. code-block:: python
from pyairtable.orm import Model, fields
from your_app.config import get_secret
class Contact(Model):
class Meta:
base_id = "appaPqizdsNHDvlEm"
table_name = "Contact"
@staticmethod
def api_key():
return get_secret("AIRTABLE_API_KEY")
first_name = fields.TextField("First Name")
age = fields.IntegerField("Age")
"""
#: The Airtable record ID for this instance. If empty, the instance
#: has never been saved to the API.
id: str = ""
#: The time when the Airtable record was created. If empty, the instance
#: has never been saved to (or fetched from) the API.
created_time: Optional[datetime.datetime] = None
#: The number of comments on this record. Only populated if the record was
#: fetched with ``count_comments=True``.
comment_count: Optional[int] = None
#: A wrapper allowing type-annotated access to ORM configuration.
meta: ClassVar["_Meta"]
_deleted: bool = False
_fetched: bool = False
_fields: Dict[FieldName, Any]
_changed: Dict[FieldName, bool]
_memoized: ClassVar[Dict[RecordId, SelfType]]
def __init_subclass__(cls, **kwargs: Any):
cls.meta = _Meta(cls)
cls._memoized = {}
cls._validate_class()
super().__init_subclass__(**kwargs)
@classmethod
def _validate_class(cls) -> None:
# Verify required Meta attributes were set (but don't call any callables)
assert cls.meta.get("api_key", required=True, call=False)
assert cls.meta.get("base_id", required=True, call=False)
assert cls.meta.get("table_name", required=True, call=False)
model_attributes = [a for a in cls.__dict__.keys() if not a.startswith("__")]
overridden = set(model_attributes).intersection(Model.__dict__.keys())
if overridden:
raise ValueError(
"Class {cls} fields clash with existing method: {name}".format(
cls=cls.__name__, name=overridden
)
)
@classmethod
def _attribute_descriptor_map(cls) -> Dict[str, AnyField]:
"""
Build a mapping of the model's attribute names to field descriptor instances.
>>> class Test(Model):
... first_name = TextField("First Name")
... age = NumberField("Age")
...
>>> Test._attribute_descriptor_map()
>>> {
... "field_name": <TextField field_name="First Name">,
... "another_Field": <NumberField field_name="Age">,
... }
"""
return {k: v for k, v in cls.__dict__.items() if isinstance(v, Field)}
@classmethod
def _field_name_descriptor_map(cls) -> Dict[FieldName, AnyField]:
"""
Build a mapping of the model's field names to field descriptor instances.
>>> class Test(Model):
... first_name = TextField("First Name")
... age = NumberField("Age")
...
>>> Test._field_name_descriptor_map()
>>> {
... "First Name": <TextField field_name="First Name">,
... "Age": <NumberField field_name="Age">,
... }
"""
return {f.field_name: f for f in cls._attribute_descriptor_map().values()}
[docs] def __init__(self, **fields: Any):
"""
Construct a model instance with field values based on the given keyword args.
>>> Contact(name="Alice", birthday=date(1980, 1, 1))
<unsaved Contact>
The keyword argument ``id=`` special-cased and sets the record ID, not a field value.
>>> Contact(id="recWPqD9izdsNvlE", name="Bob")
<Contact id='recWPqD9izdsNvlE'>
"""
try:
self.id = fields.pop("id")
except KeyError:
pass
# Field values in internal (not API) representation
self._fields = {}
# Call __set__ on each field to set field values
for key, value in fields.items():
if key not in self._attribute_descriptor_map():
raise AttributeError(key)
setattr(self, key, value)
# Only start tracking changes after the object is created
self._changed = {}
def __repr__(self) -> str:
if not self.id:
return f"<unsaved {self.__class__.__name__}>"
return f"<{self.__class__.__name__} id={self.id!r}>"
[docs] def exists(self) -> bool:
"""
Whether the instance has been saved to Airtable already.
"""
return bool(self.id)
[docs] def save(self, *, force: bool = False) -> "SaveResult":
"""
Save the model to the API.
If the instance does not exist already, it will be created;
otherwise, the existing record will be updated, using only the
fields which have been modified since it was retrieved.
Args:
force: If ``True``, all fields will be saved, even if they have not changed.
"""
if self._deleted:
raise RuntimeError(f"{self.id} was deleted")
field_values = self.to_record(only_writable=True)["fields"]
if not self.id:
record = self.meta.table.create(
field_values,
typecast=self.meta.typecast,
use_field_ids=self.meta.use_field_ids,
)
self.id = record["id"]
self.created_time = datetime_from_iso_str(record["createdTime"])
self._changed.clear()
return SaveResult(self.id, created=True, field_names=set(field_values))
if not force:
if not self._changed:
return SaveResult(self.id)
field_values = {
field_name: value
for field_name, value in field_values.items()
if self._changed.get(field_name)
}
self.meta.table.update(
self.id,
field_values,
typecast=self.meta.typecast,
use_field_ids=self.meta.use_field_ids,
)
self._changed.clear()
return SaveResult(
self.id, forced=force, updated=True, field_names=set(field_values)
)
[docs] def delete(self) -> bool:
"""
Delete the record.
Raises:
ValueError: if the record does not exist.
"""
if not self.id:
raise ValueError("cannot be deleted because it does not have id")
table = self.meta.table
result = table.delete(self.id)
self._deleted = True
# Is it even possible to get "deleted" False?
return bool(result["deleted"])
[docs] @classmethod
def all(cls, *, memoize: Optional[bool] = None, **kwargs: Any) -> List[SelfType]:
"""
Retrieve all records for this model. For all supported
keyword arguments, see :meth:`Table.all <pyairtable.Table.all>`.
Args:
memoize: |kwarg_orm_memoize|
"""
kwargs.update(cls.meta.request_kwargs)
return [
cls.from_record(record, memoize=memoize)
for record in cls.meta.table.all(**kwargs)
]
[docs] @classmethod
def first(
cls, *, memoize: Optional[bool] = None, **kwargs: Any
) -> Optional[SelfType]:
"""
Retrieve the first record for this model. For all supported
keyword arguments, see :meth:`Table.first <pyairtable.Table.first>`.
Args:
memoize: |kwarg_orm_memoize|
"""
kwargs.update(cls.meta.request_kwargs)
if record := cls.meta.table.first(**kwargs):
return cls.from_record(record, memoize=memoize)
return None
@classmethod
def _maybe_memoize(cls, instance: SelfType, memoize: Optional[bool]) -> None:
"""
If memoization is enabled, save the instance to the memoization cache.
"""
memoize = cls.meta.memoize if memoize is None else memoize
if memoize:
cls._memoized[instance.id] = instance
[docs] def to_record(self, only_writable: bool = False) -> RecordDict:
"""
Build a :class:`~pyairtable.api.types.RecordDict` to represent this instance.
This method converts internal field values into values expected by Airtable.
For example, a ``datetime`` value from :class:`~pyairtable.orm.fields.DatetimeField`
is converted into an ISO 8601 string.
Args:
only_writable: If ``True``, the result will exclude any
values which are associated with readonly fields.
"""
map_ = self._field_name_descriptor_map()
fields = {
field: None if value is None else map_[field].to_record_value(value)
for field, value in self._fields.items()
if not (map_[field].readonly and only_writable)
}
ct = datetime_to_iso_str(self.created_time) if self.created_time else ""
return {"id": self.id, "createdTime": ct, "fields": fields}
[docs] @classmethod
def from_record(
cls, record: RecordDict, *, memoize: Optional[bool] = None
) -> SelfType:
"""
Create an instance from a record dict.
Args:
record: The record data from the Airtable API.
memoize: |kwarg_orm_memoize|
"""
name_field_map = cls._field_name_descriptor_map()
# Convert Column Names into model field names
field_values = {
# Use field's to_internal_value to cast into model fields
field: (
name_field_map[field].to_internal_value(value)
if value is not None
else None
)
for (field, value) in record["fields"].items()
# Silently proceed if Airtable returns fields we don't recognize
if field in name_field_map
}
# Since instance(**field_values) will perform validation and fail on
# any readonly fields, instead we directly set instance._fields.
instance = cls(id=record["id"])
instance._fields = field_values
instance._fetched = True
instance.created_time = datetime_from_iso_str(record["createdTime"])
instance.comment_count = record.get("commentCount")
cls._maybe_memoize(instance, memoize)
return instance
[docs] @classmethod
def from_id(
cls,
record_id: RecordId,
*,
fetch: bool = True,
memoize: Optional[bool] = None,
) -> SelfType:
"""
Create an instance from a record ID.
Args:
record_id: |arg_record_id|
fetch: |kwarg_orm_fetch|
memoize: |kwarg_orm_memoize|
"""
try:
instance = cast(SelfType, cls._memoized[record_id]) # type: ignore[redundant-cast]
except KeyError:
instance = cls(id=record_id)
if fetch and not instance._fetched:
instance.fetch()
cls._maybe_memoize(instance, memoize)
return instance
[docs] def fetch(self) -> None:
"""
Fetch field values from the API and resets instance field values.
"""
if not self.id:
raise ValueError("cannot be fetched because instance does not have an id")
record = self.meta.table.get(self.id, **self.meta.request_kwargs)
unused = self.from_record(record, memoize=False)
self._fields = unused._fields
self._changed.clear()
self._fetched = True
self.created_time = unused.created_time
[docs] @classmethod
def from_ids(
cls,
record_ids: Iterable[RecordId],
*,
fetch: bool = True,
memoize: Optional[bool] = None,
) -> List[SelfType]:
"""
Create a list of instances from record IDs. If any record IDs returned
are invalid this will raise a KeyError, but only *after* retrieving all
other valid records from the API.
Args:
record_ids: |arg_record_id|
fetch: |kwarg_orm_fetch|
memoize: |kwarg_orm_memoize|
"""
if not fetch:
return [cls.from_id(record_id, fetch=False) for record_id in record_ids]
record_ids = list(record_ids)
by_id: Dict[RecordId, SelfType] = {}
if cls._memoized:
for record_id in record_ids:
try:
by_id[record_id] = cast(SelfType, cls._memoized[record_id]) # type: ignore[redundant-cast]
except KeyError:
pass
if remaining := sorted(set(record_ids) - set(by_id)):
# Only retrieve records that aren't already memoized
formula = OR(EQ(RECORD_ID(), record_id) for record_id in sorted(remaining))
by_id.update(
{obj.id: obj for obj in cls.all(formula=formula, memoize=memoize)}
)
# Ensure we return records in the same order, and raise KeyError if any are missing
return [by_id[record_id] for record_id in record_ids]
[docs] @classmethod
def batch_save(cls, models: List[SelfType]) -> None:
"""
Save a list of model instances to the Airtable API with as few
network requests as possible. Can accept a mixture of new records
(which have not been saved yet) and existing records that have IDs.
"""
if not all(isinstance(model, cls) for model in models):
raise TypeError(set(type(model) for model in models))
create_models = [model for model in models if not model.id]
update_models = [model for model in models if model.id]
create_records: List[WritableFields] = [
record["fields"]
for model in create_models
if (record := model.to_record(only_writable=True))
]
update_records: List[UpdateRecordDict] = [
{"id": record["id"], "fields": record["fields"]}
for model in update_models
if (record := model.to_record(only_writable=True))
]
if update_records:
cls.meta.table.batch_update(
update_records,
typecast=cls.meta.typecast,
use_field_ids=cls.meta.use_field_ids,
)
if create_records:
created_records = cls.meta.table.batch_create(
create_records,
typecast=cls.meta.typecast,
use_field_ids=cls.meta.use_field_ids,
)
for model, record in zip(create_models, created_records):
model.id = record["id"]
model.created_time = datetime_from_iso_str(record["createdTime"])
[docs] @classmethod
def batch_delete(cls, models: List[SelfType]) -> None:
"""
Delete a list of model instances from Airtable.
Raises:
ValueError: if the model has not been saved to Airtable.
"""
if not all(model.id for model in models):
raise ValueError("cannot delete an unsaved model")
if not all(isinstance(model, cls) for model in models):
raise TypeError(set(type(model) for model in models))
cls.meta.table.batch_delete([model.id for model in models])
@dataclass
class _Meta:
"""
Wrapper around a Model.Meta class that provides easier, typed access to
configuration values (which may or may not be defined in the original class).
"""
model: Type[Model]
@property
def _config(self) -> Mapping[str, Any]:
if not (meta := getattr(self.model, "Meta", None)):
raise AttributeError(f"{self.model.__name__}.Meta must be defined")
if isinstance(meta, dict):
return meta
try:
return cast(Mapping[str, Any], meta.__dict__)
except AttributeError:
raise TypeError(
f"{self.model.__name__}.Meta must be a dict or class; got {type(meta)}"
)
def get(
self,
name: str,
default: Any = None,
required: bool = False,
call: bool = True,
check_types: Optional["_ClassInfo"] = None,
) -> Any:
"""
Given a name, retrieve the model configuration with that name.
Args:
default: The default value to use if the name is not defined.
required: If ``True``, raises ``ValueError`` if the name is undefined or None.
call: If ``False``, does not execute any callables to retrieve this value;
it will consider the callable itself as the value.
check_types: If set, will raise a ``TypeError`` if the value is not
an instance of the given type(s).
"""
if required and name not in self._config:
raise ValueError(f"{self.model.__name__}.Meta.{name} must be defined")
value = self._config.get(name, default)
if callable(value) and call:
value = value()
if required and value is None:
raise ValueError(f"{self.model.__name__}.Meta.{name} cannot be None")
if check_types is not None and not isinstance(value, check_types):
raise TypeError(f"expected {check_types!r}; got {type(value)}")
return value
@property
def api_key(self) -> str:
return str(self.get("api_key", required=True))
@property
def timeout(self) -> Optional[TimeoutTuple]:
return self.get( # type: ignore[no-any-return]
"timeout",
default=None,
check_types=(type(None), tuple),
)
@property
def retry_strategy(self) -> Optional[Union[bool, retrying.Retry]]:
return self.get( # type: ignore[no-any-return]
"retry",
default=True,
check_types=(type(None), bool, retrying.Retry),
)
@cached_property
def api(self) -> Api:
return Api(
self.api_key,
timeout=self.timeout,
retry_strategy=self.retry_strategy,
)
@property
def base_id(self) -> str:
return str(self.get("base_id", required=True))
@property
def base(self) -> Base:
return self.api.base(self.base_id)
@property
def table_name(self) -> str:
return str(self.get("table_name", required=True))
@property
def table(self) -> Table:
return self.base.table(self.table_name)
@property
def typecast(self) -> bool:
return bool(self.get("typecast", default=True))
@property
def use_field_ids(self) -> bool:
return bool(self.get("use_field_ids", default=False))
@property
def memoize(self) -> bool:
return bool(self.get("memoize", default=False))
@property
def request_kwargs(self) -> Dict[str, Any]:
return {
"user_locale": None,
"cell_format": "json",
"time_zone": None,
"use_field_ids": self.use_field_ids,
}
[docs]@dataclass(frozen=True)
class SaveResult:
"""
Represents the result of saving a record to the API. The result's
attributes contain more granular information about the save operation:
>>> result = model.save()
>>> result.record_id
'recWPqD9izdsNvlE'
>>> result.created
False
>>> result.updated
True
>>> result.forced
False
>>> result.field_names
{'Name', 'Email'}
If none of the model's fields have changed, calling :meth:`~pyairtable.orm.Model.save`
will not perform any API requests and will return a SaveResult with no changes.
>>> model = YourModel()
>>> result = model.save()
>>> result.saved
True
>>> second_result = model.save()
>>> second_result.saved
False
For backwards compatibility, instances of SaveResult will evaluate as truthy
if the record was created, and falsy if the record was not created.
"""
record_id: RecordId
created: bool = False
updated: bool = False
forced: bool = False
field_names: Set[FieldName] = dataclasses.field(default_factory=set)
def __bool__(self) -> bool:
"""
Returns ``True`` if the record was created. This is for backwards compatibility
with the behavior of :meth:`~pyairtable.orm.Model.save` prior to the 3.0 release,
which returned a boolean indicating whether a record was created.
"""
warnings.warn(
"Model.save() now returns SaveResult instead of bool; switch"
" to checking Model.save().created instead before the 4.0 release.",
DeprecationWarning,
stacklevel=2,
)
return self.created
@property
def saved(self) -> bool:
"""
Whether the record was saved to the API. If ``False``, this indicates there
were no changes to the model and the :meth:`~pyairtable.orm.Model.save`
operation was not forced.
"""
return self.created or self.updated