Add a direct access to tables, for stronger typing.

airtable/airtable.py

@@ -159,3 +159,34 @@ def delete(self, table_name, record_id):
         if check_string(table_name) and check_string(record_id):
             url = posixpath.join(table_name, record_id)
             return self.__request('DELETE', url)
+
+    def table(self, table_name):
+        return AirtableTable(self, table_name)
+
+
+class AirtableTable(object):
+    def __init__(self, base_id, table_name, *args, **kwargs):
+        """Create a client to connect to an Airtable Table.
+
+        Args:
+            - base_id: The ID of the base, e.g. "appA0CDAE34F"
+            - table_name: The name or ID of the table, e.g. "tbl123adfe4"
+            - api_key: The API secret key, e.g. "keyBAAE123C"
+            - dict_class: the class to use to build dictionaries for returning
+                  fields. By default the fields are kept in the order they were
+                  returned by the API using an OrderedDict, but you can switch
+                  to a simple dict if you prefer.
+        """
+        self.table_name = table_name
+        if isinstance(base_id, Airtable):
+            self._client = base_id
+            return
+        self._client = Airtable(base_id, *args, **kwargs)
+
+    def __getattr__(self, name):
+        if name not in set(['get', 'create', 'iterate', 'update', 'update_all', 'delete']):
+            raise AttributeError(
+                "'%s' object has no attribute '%s'" % (self.__class__.__name__, name))
+        return lambda *args, **kwargs: getattr(self._client, name)(
+            self.table_name, *args, **kwargs)
+

airtable/airtable.pyi

@@ -1,4 +1,5 @@
 import typing
+from typing import Any, TypedDict
 
 API_URL: str
 API_VERSION: str
@@ -9,45 +10,51 @@ class IsNotInteger(Exception):
 class IsNotString(Exception):
     ...
 
-def check_integer(n: typing.Any) -> bool:
+def check_integer(n: Any) -> bool:
     ...
 
-def check_string(s: typing.Any) -> bool:
+def check_string(s: Any) -> bool:
     ...
 
-def create_payload(data: typing.Dict[str, typing.Any]) \
-        -> typing.Dict[str, typing.Dict[str, typing.Any]]:
+def create_payload(data: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
     ...
 
-_Record = typing.Dict[str, typing.Union[str, typing.Dict[str, typing.Any]]]
 
-class Airtable(object):
-    airtable_url: str = ...
-    base_url: str = ...
-    headers: typing.Dict[str, str] = ...
+_DefaultRecord = typing.Dict[str, typing.Any]
+RecordType = typing.TypeVar('RecordType', bound=_DefaultRecord)
+
+class _Record(TypedDict, Generic[RecordType]):
+    id: str
+    createdTime: str
+    fields: RecordType
+
+
+class _DeletedRecord(TypedDict):
+    deleted: bool
+    id: str
+
 
+class AirtableTable(Generic[RecordType]):
     def __init__(self, base_id: str, api_key: str) -> None:
         ...
 
     def iterate(
             self,
-            table_name: str,
             batch_size: int = 0,
             filter_by_formula: typing.Optional[str] = None,
-            view: typing.Optional[str] = None) -> typing.Iterator[_Record]:
+            view: typing.Optional[str] = None) -> typing.Iterator[_Record[RecordType]]:
         ...
 
     @typing.overload
     def get(
             self,
-            table_name: str,
             record_id: None = None,
             limit: int = 0,
             offset: typing.Optional[int] = None,
             filter_by_formula: typing.Optional[str] = None,
             view: typing.Optional[str] = None) \
-            -> typing.Dict[str, typing.List[_Record]]:
-     ...
+            -> typing.Dict[str, typing.List[_Record[RecordType]]]:
+        ...
 
     @typing.overload
     def get(
@@ -58,17 +65,75 @@ class Airtable(object):
             offset: typing.Optional[int] = None,
             filter_by_formula: typing.Optional[str] = None,
             view: typing.Optional[str] = None) \
-            -> _Record:
+            -> _Record[RecordType]:
+        ...
+
+    def create(self, data: typing.Dict[str, typing.Any]) -> _Record[RecordType]:
+        ...
+
+    def update(self, record_id: str, data: RecordType) -> _Record[RecordType]:
+        ...
+
+    def update_all(self, record_id: str, data: RecordType) -> _Record[RecordType]:
+        ...
+
+    def delete(self, table_name: str, record_id: str) -> _DeletedRecord:
+        ...
+
+
+class Airtable(object):
+    airtable_url: str = ...
+    base_url: str = ...
+    headers: Dict[str, str] = ...
+
+    def __init__(self, base_id: str, api_key: str) -> None:
         ...
 
-    def create(self, table_name: str, data: typing.Dict[str, typing.Any]) -> _Record:
+    def iterate(
+            self,
+            table_name: str,
+            batch_size: int = 0,
+            filter_by_formula: Optional[str] = None,
+            view: Optional[str] = None) -> Iterator[_Record]:
         ...
-
-    def update(self, table_name: str, record_id: str, data: typing.Dict[str, typing.Any]) -> _Record:
+
+    @overload
+    def get(
+            self,
+            table_name: str,
+            record_id: None = None,
+            limit: int = 0,
+            offset: Optional[int] = None,
+            filter_by_formula: Optional[str] = None,
+            view: Optional[str] = None) \
+            -> Dict[str, List[_Record]]:
+     ...
+
+    @overload
+    def get(
+            self,
+            table_name: str,
+            record_id: str,
+            limit: int = 0,
+            offset: Optional[int] = None,
+            filter_by_formula: Optional[str] = None,
+            view: Optional[str] = None) \
+            -> _Record[_DefaultRecord]:
+        ...
+
+    def create(self, table_name: str, data: _DefaultRecord) -> _Record[_DefaultRecord]:
+        ...
+
+    def update(self, table_name: str, record_id: str, data: _DefaultRecord) \
+            -> _Record[_DefaultRecord]:
+        ...
+
+    def update_all(self, table_name: str, record_id: str, data: _DefaultRecord) \
+            -> _Record[_DefaultRecord]:
         ...
 
-    def update_all(self, table_name: str, record_id: str, data: typing.Dict[str, typing.Any]) -> _Record:
+    def delete(self, table_name: str, record_id: str) -> _DeletedRecord:
         ...
 
-    def delete(self, table_name: str, record_id: str) -> typing.Dict[str, typing.Union[str, bool]]:
+    def table(self, table_name: str) -> AirtableTable[RecordType]:
         ...