0
# Record Operations
1
2
Advanced record handling capabilities including batch operations, upserts, pagination, filtering, and sorting. These operations provide efficient data management for single records and bulk operations.
3
4
## Capabilities
5
6
### Single Record Operations
7
8
Basic CRUD operations for individual records with full control over field updates and retrieval options.
9
10
```python { .api }
11
def get(self, record_id: str, **options) -> dict:
12
"""
13
Retrieve single record by ID.
14
15
Parameters:
16
- record_id: Record ID (starts with 'rec')
17
- cell_format: 'json' (default) or 'string'
18
- time_zone: Timezone for datetime fields (e.g., 'America/New_York')
19
- user_locale: Locale for formatting (e.g., 'en-US')
20
- use_field_ids: Return field IDs instead of names
21
22
Returns:
23
Record dict with 'id', 'createdTime', and 'fields' keys
24
"""
25
26
def create(self, fields: dict, typecast: bool = False,
27
use_field_ids: Optional[bool] = None) -> dict:
28
"""
29
Create single record.
30
31
Parameters:
32
- fields: Field name/ID -> value mapping
33
- typecast: Automatically convert field values to correct types
34
- use_field_ids: Return field IDs in response
35
36
Returns:
37
Created record dict
38
"""
39
40
def update(self, record_id: str, fields: dict,
41
replace: bool = False, typecast: bool = False,
42
use_field_ids: Optional[bool] = None) -> dict:
43
"""
44
Update single record.
45
46
Parameters:
47
- record_id: Record ID to update
48
- fields: Field updates (name/ID -> value mapping)
49
- replace: If True, replace all fields (PUT). If False, merge (PATCH)
50
- typecast: Automatically convert field values
51
- use_field_ids: Return field IDs in response
52
53
Returns:
54
Updated record dict
55
"""
56
57
def delete(self, record_id: str) -> dict:
58
"""
59
Delete single record.
60
61
Parameters:
62
- record_id: Record ID to delete
63
64
Returns:
65
Dict with 'id' and 'deleted': True
66
"""
67
```
68
69
### Batch Record Operations
70
71
Efficient bulk operations for creating, updating, and deleting multiple records in single API calls.
72
73
```python { .api }
74
def batch_create(self, records: list[dict], typecast: bool = False,
75
use_field_ids: Optional[bool] = None) -> list[dict]:
76
"""
77
Create multiple records in batches.
78
79
Parameters:
80
- records: List of field dicts to create
81
- typecast: Automatically convert field values
82
- use_field_ids: Return field IDs in response
83
84
Returns:
85
List of created record dicts
86
"""
87
88
def batch_update(self, records: list[dict], replace: bool = False,
89
typecast: bool = False,
90
use_field_ids: Optional[bool] = None) -> list[dict]:
91
"""
92
Update multiple records in batches.
93
94
Parameters:
95
- records: List of dicts with 'id' and 'fields' keys
96
- replace: If True, replace all fields (PUT). If False, merge (PATCH)
97
- typecast: Automatically convert field values
98
- use_field_ids: Return field IDs in response
99
100
Returns:
101
List of updated record dicts
102
"""
103
104
def batch_delete(self, record_ids: list[str]) -> list[dict]:
105
"""
106
Delete multiple records in batches.
107
108
Parameters:
109
- record_ids: List of record IDs to delete
110
111
Returns:
112
List of dicts with 'id' and 'deleted': True
113
"""
114
115
def batch_upsert(self, records: list[dict], key_fields: list[str],
116
replace: bool = False, typecast: bool = False,
117
use_field_ids: Optional[bool] = None) -> dict:
118
"""
119
Create or update records based on key fields.
120
121
Parameters:
122
- records: List of record dicts (with optional 'id' field)
123
- key_fields: Field names to match existing records
124
- replace: If True, replace all fields. If False, merge
125
- typecast: Automatically convert field values
126
- use_field_ids: Return field IDs in response
127
128
Returns:
129
Dict with 'createdRecords', 'updatedRecords', and 'records' lists
130
"""
131
```
132
133
### Record Retrieval and Pagination
134
135
Flexible record retrieval with filtering, sorting, pagination, and field selection options.
136
137
```python { .api }
138
def all(self, **options) -> list[dict]:
139
"""
140
Retrieve all matching records.
141
142
Parameters:
143
- view: View name or ID to filter records
144
- page_size: Records per page (1-100, default 100)
145
- max_records: Maximum total records to return
146
- fields: List of field names/IDs to include
147
- sort: List of sort specifications
148
- formula: Formula string to filter records
149
- cell_format: 'json' (default) or 'string'
150
- time_zone: Timezone for datetime fields
151
- user_locale: Locale for formatting
152
- use_field_ids: Return field IDs instead of names
153
154
Returns:
155
List of all matching record dicts
156
"""
157
158
def iterate(self, **options) -> Iterator[list[dict]]:
159
"""
160
Iterate through records in pages.
161
162
Parameters: Same as all()
163
164
Yields:
165
List of records for each page
166
"""
167
168
def first(self, **options) -> Optional[dict]:
169
"""
170
Get first matching record.
171
172
Parameters: Same as all() (automatically sets max_records=1)
173
174
Returns:
175
First matching record dict or None
176
"""
177
```
178
179
### Usage Examples
180
181
#### Single Record Operations
182
183
```python
184
from pyairtable import Api
185
186
api = Api('your_token')
187
table = api.table('base_id', 'table_name')
188
189
# Create record
190
record = table.create({
191
'Name': 'John Doe',
192
'Email': 'john@example.com',
193
'Age': 30
194
})
195
print(f"Created: {record['id']}")
196
197
# Get record
198
retrieved = table.get(record['id'])
199
print(f"Name: {retrieved['fields']['Name']}")
200
201
# Update record (partial update)
202
updated = table.update(record['id'], {
203
'Age': 31,
204
'Status': 'Active'
205
})
206
207
# Replace record (full replacement)
208
replaced = table.update(record['id'], {
209
'Name': 'Jane Doe',
210
'Email': 'jane@example.com'
211
}, replace=True)
212
213
# Delete record
214
deleted = table.delete(record['id'])
215
print(f"Deleted: {deleted['deleted']}")
216
```
217
218
#### Batch Operations
219
220
```python
221
# Batch create multiple records
222
records_to_create = [
223
{'Name': 'Alice', 'Department': 'Engineering'},
224
{'Name': 'Bob', 'Department': 'Sales'},
225
{'Name': 'Carol', 'Department': 'Marketing'}
226
]
227
228
created_records = table.batch_create(records_to_create)
229
print(f"Created {len(created_records)} records")
230
231
# Batch update
232
updates = [
233
{'id': created_records[0]['id'], 'fields': {'Status': 'Active'}},
234
{'id': created_records[1]['id'], 'fields': {'Status': 'Pending'}}
235
]
236
237
updated_records = table.batch_update(updates)
238
239
# Batch upsert (create or update based on Name field)
240
upsert_data = [
241
{'Name': 'Alice', 'Department': 'Senior Engineering'}, # Updates existing
242
{'Name': 'David', 'Department': 'HR'} # Creates new
243
]
244
245
result = table.batch_upsert(upsert_data, key_fields=['Name'])
246
print(f"Created: {len(result['createdRecords'])}, Updated: {len(result['updatedRecords'])}")
247
248
# Batch delete
249
record_ids = [r['id'] for r in created_records]
250
deleted = table.batch_delete(record_ids)
251
print(f"Deleted {len(deleted)} records")
252
```
253
254
#### Advanced Retrieval
255
256
```python
257
# Get all records with filtering and sorting
258
records = table.all(
259
view='Active Users',
260
fields=['Name', 'Email', 'Department'],
261
sort=[
262
{'field': 'Department', 'direction': 'asc'},
263
{'field': 'Name', 'direction': 'asc'}
264
],
265
max_records=50
266
)
267
268
# Use formula for complex filtering
269
from pyairtable.formulas import match, AND, OR
270
271
# Records where Department is 'Engineering' and Status is 'Active'
272
engineering_active = table.all(
273
formula=match({
274
'Department': 'Engineering',
275
'Status': 'Active'
276
})
277
)
278
279
# More complex formula
280
complex_filter = OR(
281
AND(match({'Department': 'Engineering'}), match({'Level': ('>=', 3)})),
282
match({'Department': 'Management'})
283
)
284
records = table.all(formula=complex_filter)
285
286
# Paginate through large datasets
287
for page in table.iterate(page_size=50):
288
print(f"Processing {len(page)} records")
289
for record in page:
290
process_record(record)
291
292
# Get first matching record
293
first_user = table.first(
294
formula=match({'Status': 'New'}),
295
sort=[{'field': 'Created', 'direction': 'asc'}]
296
)
297
if first_user:
298
print(f"Oldest new user: {first_user['fields']['Name']}")
299
```
300
301
#### Type Casting and Field Options
302
303
```python
304
# Enable automatic type conversion
305
record = table.create({
306
'Name': 'Test User',
307
'Age': '25', # String will be converted to number
308
'Active': 'true', # String will be converted to boolean
309
'Join Date': '2023-01-15' # String will be converted to date
310
}, typecast=True)
311
312
# Use field IDs instead of names
313
api_with_field_ids = Api('your_token', use_field_ids=True)
314
table_with_ids = api_with_field_ids.table('base_id', 'table_name')
315
316
record = table_with_ids.create({
317
'fld1234567890abcd': 'Value using field ID'
318
})
319
320
# Custom formatting options
321
record = table.get('record_id',
322
cell_format='string', # Get formatted strings instead of raw values
323
time_zone='America/New_York', # Convert dates to specific timezone
324
user_locale='en-US' # Use specific locale for formatting
325
)
326
```
327
328
#### Error Handling for Batch Operations
329
330
```python
331
import requests
332
333
try:
334
# Batch operations can partially succeed
335
result = table.batch_create(large_dataset)
336
except requests.exceptions.HTTPError as e:
337
# Check if it's a rate limit or validation error
338
if e.response.status_code == 429:
339
print("Rate limited, wait and retry")
340
elif e.response.status_code == 422:
341
print("Validation error in data")
342
error_details = e.response.json()
343
print(f"Error: {error_details}")
344
```