0
# Actions and Lifecycle Hooks
1
2
Action decorators for bulk operations and lifecycle hooks for customizing create, update, and delete operations. The action system provides comprehensive support for bulk operations on model views with custom processing logic and user interface integration.
3
4
## Capabilities
5
6
### Action Decorator
7
8
Decorator for exposing bulk actions on ModelView classes, enabling custom operations on multiple selected records with confirmation dialogs and custom processing logic.
9
10
```python { .api }
11
from flask_appbuilder.actions import action
12
13
@action(name, text, confirmation=None, icon=None, multiple=True, single=True)
14
def action_function(self, items):
15
"""
16
Decorator to expose bulk actions on ModelView.
17
18
Parameters:
19
- name: Unique action name (used in URLs and forms)
20
- text: Display text for action button/menu
21
- confirmation: Confirmation message (None for no confirmation)
22
- icon: Font Awesome icon class (e.g., "fa-check")
23
- multiple: Show action on list view for multiple selection
24
- single: Show action on show view for single item
25
26
Function Parameters:
27
- items: List of model instances to process
28
29
Returns:
30
Flask response (redirect, render template, etc.)
31
"""
32
33
# Basic action example
34
class PersonModelView(ModelView):
35
datamodel = SQLAInterface(Person)
36
37
@action("activate", "Activate", "Activate selected persons?", "fa-check")
38
def activate_persons(self, items):
39
"""Activate selected persons."""
40
if not items:
41
flash("No items selected", "warning")
42
return redirect(self.get_redirect())
43
44
count = 0
45
for item in items:
46
if not item.active:
47
item.active = True
48
self.datamodel.edit(item)
49
count += 1
50
51
flash(f"Activated {count} persons", "success")
52
return redirect(self.get_redirect())
53
54
@action("deactivate", "Deactivate", "Deactivate selected persons?", "fa-times")
55
def deactivate_persons(self, items):
56
"""Deactivate selected persons."""
57
count = 0
58
for item in items:
59
if item.active:
60
item.active = False
61
self.datamodel.edit(item)
62
count += 1
63
64
flash(f"Deactivated {count} persons", "success")
65
return redirect(self.get_redirect())
66
67
# Action with complex processing
68
@action("export", "Export to CSV", icon="fa-download", multiple=True, single=False)
69
def export_csv(self, items):
70
"""Export selected items to CSV file."""
71
import csv
72
from flask import make_response
73
from io import StringIO
74
75
output = StringIO()
76
writer = csv.writer(output)
77
78
# Write header
79
writer.writerow(['Name', 'Email', 'Department', 'Active'])
80
81
# Write data
82
for item in items:
83
writer.writerow([
84
item.name,
85
item.email,
86
item.department.name if item.department else '',
87
'Yes' if item.active else 'No'
88
])
89
90
# Create response
91
response = make_response(output.getvalue())
92
response.headers['Content-Type'] = 'text/csv'
93
response.headers['Content-Disposition'] = 'attachment; filename=persons.csv'
94
95
return response
96
97
# Action with email notifications
98
@action("notify", "Send Notification", "Send email to selected persons?", "fa-envelope")
99
def send_notification(self, items):
100
"""Send email notification to selected persons."""
101
if not items:
102
flash("No items selected", "warning")
103
return redirect(self.get_redirect())
104
105
# Get notification message from form or session
106
message = request.form.get('notification_message', 'Default notification')
107
108
success_count = 0
109
error_count = 0
110
111
for item in items:
112
try:
113
send_email(
114
to=item.email,
115
subject="Important Notification",
116
body=message
117
)
118
success_count += 1
119
except Exception as e:
120
error_count += 1
121
logger.error(f"Failed to send email to {item.email}: {e}")
122
123
flash(f"Sent {success_count} notifications, {error_count} failed",
124
"success" if error_count == 0 else "warning")
125
126
return redirect(self.get_redirect())
127
128
# Single-item action (show view only)
129
@action("promote", "Promote to Manager", "Promote this person to manager?",
130
"fa-arrow-up", multiple=False, single=True)
131
def promote_to_manager(self, items):
132
"""Promote single person to manager role."""
133
item = items[0] # Single item
134
135
# Business logic validation
136
if item.department is None:
137
flash("Cannot promote person without department", "error")
138
return redirect(self.get_redirect())
139
140
if item.years_experience < 3:
141
flash("Minimum 3 years experience required for promotion", "error")
142
return redirect(self.get_redirect())
143
144
# Update role
145
manager_role = db.session.query(Role).filter_by(name='Manager').first()
146
if manager_role:
147
item.role = manager_role
148
item.promotion_date = datetime.date.today()
149
self.datamodel.edit(item)
150
151
flash(f"Promoted {item.name} to Manager", "success")
152
153
# Send notification
154
send_promotion_notification(item)
155
156
return redirect(self.get_redirect())
157
```
158
159
### ActionItem Class
160
161
Class representing individual actions with properties for display, behavior, and execution configuration.
162
163
```python { .api }
164
from flask_appbuilder.actions import ActionItem
165
166
class ActionItem(object):
167
"""
168
Class representing a single action with configuration.
169
"""
170
171
def __init__(self, name, text, confirmation=None, icon=None,
172
multiple=True, single=True, func=None):
173
"""
174
Initialize action item.
175
176
Parameters:
177
- name: Unique action identifier
178
- text: Display text for UI
179
- confirmation: Confirmation dialog message
180
- icon: Font Awesome icon class
181
- multiple: Available on list view (multiple selection)
182
- single: Available on show view (single item)
183
- func: Action function to execute
184
"""
185
self.name = name
186
self.text = text
187
self.confirmation = confirmation
188
self.icon = icon
189
self.multiple = multiple
190
self.single = single
191
self.func = func
192
193
# Properties and methods
194
name = "" # Action identifier
195
text = "" # Display text
196
confirmation = None # Confirmation message
197
icon = None # Font Awesome icon
198
multiple = True # Show on list view
199
single = True # Show on show view
200
func = None # Action function
201
202
# Manual action registration example
203
class PersonModelView(ModelView):
204
datamodel = SQLAInterface(Person)
205
206
def __init__(self):
207
super(PersonModelView, self).__init__()
208
209
# Manually register actions
210
self.actions = {
211
'custom_action': ActionItem(
212
name='custom_action',
213
text='Custom Action',
214
confirmation='Execute custom action?',
215
icon='fa-cog',
216
multiple=True,
217
single=True,
218
func=self.custom_action_handler
219
)
220
}
221
222
def custom_action_handler(self, items):
223
"""Custom action handler method."""
224
# Process items
225
for item in items:
226
process_custom_logic(item)
227
228
flash(f"Processed {len(items)} items", "success")
229
return redirect(self.get_redirect())
230
231
# Conditional action availability
232
class ConditionalPersonView(ModelView):
233
datamodel = SQLAInterface(Person)
234
235
@action("admin_action", "Admin Only", "Admin action?", "fa-shield",
236
multiple=True, single=True)
237
def admin_only_action(self, items):
238
"""Action only available to administrators."""
239
# Check user permissions
240
if not g.user.has_role('Admin'):
241
flash("Insufficient permissions", "error")
242
return redirect(self.get_redirect())
243
244
# Execute admin action
245
for item in items:
246
execute_admin_action(item)
247
248
return redirect(self.get_redirect())
249
250
def _get_actions_dict(self):
251
"""Override to conditionally show actions."""
252
actions = super()._get_actions_dict()
253
254
# Remove admin action for non-admin users
255
if not g.user.has_role('Admin'):
256
actions.pop('admin_action', None)
257
258
return actions
259
```
260
261
### Lifecycle Hooks
262
263
Comprehensive lifecycle hooks for customizing database operations during create, read, update, and delete operations with pre and post processing capabilities.
264
265
```python { .api }
266
# Complete set of lifecycle hooks for ModelView and ModelRestApi
267
268
class PersonModelView(ModelView):
269
datamodel = SQLAInterface(Person)
270
271
# CREATE hooks
272
def pre_add(self, item):
273
"""
274
Called before adding new item to database.
275
276
Parameters:
277
- item: Model instance about to be added
278
279
Use for:
280
- Validation
281
- Setting default values
282
- Data transformation
283
- Business rule enforcement
284
"""
285
# Set creation timestamp
286
item.created_on = datetime.datetime.now()
287
288
# Generate employee ID
289
if not item.employee_id:
290
item.employee_id = self.generate_employee_id()
291
292
# Validate business rules
293
if item.department and item.salary:
294
max_salary = self.get_max_salary_for_department(item.department)
295
if item.salary > max_salary:
296
flash(f"Salary exceeds department maximum of ${max_salary}", "warning")
297
298
# Set default manager based on department
299
if item.department and not item.manager:
300
item.manager = item.department.default_manager
301
302
def post_add(self, item):
303
"""
304
Called after successfully adding item to database.
305
306
Parameters:
307
- item: Added model instance (has ID assigned)
308
309
Use for:
310
- Notifications
311
- Logging
312
- Creating related records
313
- External system integration
314
"""
315
# Send welcome email
316
try:
317
send_welcome_email(item.email, item.name)
318
except Exception as e:
319
logger.error(f"Failed to send welcome email: {e}")
320
321
# Create user account
322
if item.email and not item.user_account:
323
try:
324
user = self.create_user_account(item)
325
item.user_account = user
326
self.datamodel.edit(item)
327
except Exception as e:
328
logger.error(f"Failed to create user account: {e}")
329
330
# Log the action
331
self.audit_log.log_creation(item, g.user)
332
333
# Update department statistics
334
self.update_department_stats(item.department)
335
336
# UPDATE hooks
337
def pre_update(self, item):
338
"""
339
Called before updating item in database.
340
341
Parameters:
342
- item: Model instance about to be updated
343
344
Access original values via item._sa_instance_state.committed_state
345
"""
346
# Track field changes for audit
347
self.track_changes(item)
348
349
# Update timestamp
350
item.updated_on = datetime.datetime.now()
351
item.updated_by = g.user
352
353
# Validate department change
354
if self.has_field_changed(item, 'department_id'):
355
old_dept = self.get_original_value(item, 'department_id')
356
new_dept = item.department_id
357
358
if not self.validate_department_transfer(item, old_dept, new_dept):
359
raise ValidationError("Department transfer not allowed")
360
361
# Validate salary change
362
if self.has_field_changed(item, 'salary'):
363
old_salary = self.get_original_value(item, 'salary')
364
new_salary = item.salary
365
366
if new_salary > old_salary * 1.5: # Max 50% increase
367
flash("Salary increase exceeds policy limit", "warning")
368
369
def post_update(self, item):
370
"""
371
Called after successfully updating item.
372
373
Parameters:
374
- item: Updated model instance
375
"""
376
# Send change notifications
377
if self.has_critical_changes(item):
378
send_change_notification(item, self.get_changed_fields(item))
379
380
# Update related records
381
if self.has_field_changed(item, 'department_id'):
382
self.update_department_assignments(item)
383
384
# Sync with external systems
385
try:
386
sync_with_hr_system(item)
387
except Exception as e:
388
logger.error(f"HR system sync failed: {e}")
389
390
# Log changes
391
self.audit_log.log_update(item, self.get_changed_fields(item), g.user)
392
393
# DELETE hooks
394
def pre_delete(self, item):
395
"""
396
Called before deleting item from database.
397
398
Parameters:
399
- item: Model instance about to be deleted
400
401
Use for:
402
- Validation (can item be deleted?)
403
- Cascade operations
404
- Backup/archival
405
"""
406
# Check if person can be deleted
407
if item.is_manager and item.subordinates:
408
raise ValidationError("Cannot delete manager with subordinates")
409
410
if item.active_projects:
411
raise ValidationError("Cannot delete person with active projects")
412
413
# Archive related data
414
self.archive_person_data(item)
415
416
# Backup before deletion
417
self.backup_person_record(item)
418
419
# Notify stakeholders
420
notify_person_deletion(item)
421
422
def post_delete(self, item):
423
"""
424
Called after successfully deleting item.
425
426
Parameters:
427
- item: Deleted model instance (no longer in database)
428
"""
429
# Clean up related records
430
self.cleanup_person_references(item)
431
432
# Deactivate user account
433
if item.user_account:
434
deactivate_user_account(item.user_account)
435
436
# Update statistics
437
self.update_department_stats(item.department)
438
439
# Log deletion
440
self.audit_log.log_deletion(item, g.user)
441
442
# External system cleanup
443
cleanup_external_systems(item)
444
445
# READ hooks (for data retrieval customization)
446
def pre_get(self, data):
447
"""
448
Called before returning single item data (API only).
449
450
Parameters:
451
- data: Serialized item data dict
452
453
Returns:
454
Modified data dict
455
"""
456
# Add computed fields
457
data['display_name'] = f"{data['first_name']} {data['last_name']}"
458
data['years_service'] = calculate_years_service(data['hire_date'])
459
460
# Remove sensitive data based on permissions
461
if not g.user.has_role('HR'):
462
data.pop('salary', None)
463
data.pop('performance_rating', None)
464
465
return data
466
467
def pre_get_list(self, data):
468
"""
469
Called before returning list data (API only).
470
471
Parameters:
472
- data: Dict with count, ids, result array
473
474
Returns:
475
Modified data dict
476
"""
477
# Add summary statistics
478
data['summary'] = {
479
'total_active': sum(1 for item in data['result'] if item.get('active')),
480
'total_inactive': sum(1 for item in data['result'] if not item.get('active'))
481
}
482
483
# Filter results based on user permissions
484
if not g.user.has_role('Manager'):
485
# Hide salary information for non-managers
486
for item in data['result']:
487
item.pop('salary', None)
488
489
return data
490
491
# Helper methods for hooks
492
def has_field_changed(self, item, field_name):
493
"""Check if specific field has changed."""
494
if hasattr(item, '_sa_instance_state'):
495
committed = item._sa_instance_state.committed_state
496
current_value = getattr(item, field_name)
497
original_value = committed.get(field_name)
498
return current_value != original_value
499
return False
500
501
def get_original_value(self, item, field_name):
502
"""Get original value of field before changes."""
503
if hasattr(item, '_sa_instance_state'):
504
return item._sa_instance_state.committed_state.get(field_name)
505
return None
506
507
def get_changed_fields(self, item):
508
"""Get list of fields that have changed."""
509
changed = []
510
if hasattr(item, '_sa_instance_state'):
511
committed = item._sa_instance_state.committed_state
512
for field_name in committed.keys():
513
if self.has_field_changed(item, field_name):
514
changed.append(field_name)
515
return changed
516
517
def track_changes(self, item):
518
"""Track changes for audit trail."""
519
self._changed_fields = self.get_changed_fields(item)
520
self._original_values = {}
521
522
for field in self._changed_fields:
523
self._original_values[field] = self.get_original_value(item, field)
524
```
525
526
### Custom Action Workflows
527
528
Advanced action patterns for complex business workflows, multi-step operations, and integration with external systems.
529
530
```python { .api }
531
# Multi-step action workflow
532
class OrderModelView(ModelView):
533
datamodel = SQLAInterface(Order)
534
535
@action("process_order", "Process Order", "Process selected orders?", "fa-cogs")
536
def process_order_workflow(self, items):
537
"""Multi-step order processing workflow."""
538
processed = []
539
failed = []
540
541
for order in items:
542
try:
543
# Step 1: Validate order
544
if not self.validate_order(order):
545
failed.append(f"Order {order.id}: Validation failed")
546
continue
547
548
# Step 2: Check inventory
549
if not self.check_inventory(order):
550
failed.append(f"Order {order.id}: Insufficient inventory")
551
continue
552
553
# Step 3: Process payment
554
if not self.process_payment(order):
555
failed.append(f"Order {order.id}: Payment failed")
556
continue
557
558
# Step 4: Update status
559
order.status = 'PROCESSING'
560
order.processed_date = datetime.datetime.now()
561
self.datamodel.edit(order)
562
563
# Step 5: Create shipment
564
shipment = self.create_shipment(order)
565
566
processed.append(order.id)
567
568
except Exception as e:
569
failed.append(f"Order {order.id}: {str(e)}")
570
571
# Show results
572
if processed:
573
flash(f"Successfully processed orders: {', '.join(map(str, processed))}", "success")
574
575
if failed:
576
flash(f"Failed to process: {'; '.join(failed)}", "error")
577
578
return redirect(self.get_redirect())
579
580
# Action with user input form
581
@action("assign_bulk", "Bulk Assign", icon="fa-users")
582
def bulk_assign_action(self, items):
583
"""Action that requires additional user input."""
584
if request.method == 'POST':
585
# Process form submission
586
assignee_id = request.form.get('assignee_id')
587
priority = request.form.get('priority')
588
due_date = request.form.get('due_date')
589
590
if not assignee_id:
591
flash("Please select an assignee", "error")
592
return redirect(self.get_redirect())
593
594
assignee = db.session.query(User).get(assignee_id)
595
596
for item in items:
597
item.assigned_to = assignee
598
item.priority = priority
599
item.due_date = datetime.datetime.strptime(due_date, '%Y-%m-%d').date()
600
self.datamodel.edit(item)
601
602
flash(f"Assigned {len(items)} items to {assignee.username}", "success")
603
return redirect(self.get_redirect())
604
605
else:
606
# Show form for user input
607
users = db.session.query(User).filter(User.active == True).all()
608
609
return self.render_template(
610
'bulk_assign_form.html',
611
items=items,
612
users=users,
613
action_url=url_for('PersonModelView.action', name='assign_bulk')
614
)
615
616
# Action with external API integration
617
@action("sync_external", "Sync with External System",
618
"Sync selected items with external system?", "fa-sync")
619
def sync_external_system(self, items):
620
"""Sync items with external system."""
621
import requests
622
623
success_count = 0
624
error_count = 0
625
626
for item in items:
627
try:
628
# Prepare data for external API
629
sync_data = {
630
'id': item.id,
631
'name': item.name,
632
'email': item.email,
633
'department': item.department.name if item.department else None
634
}
635
636
# Call external API
637
response = requests.post(
638
'https://api.external-system.com/sync',
639
json=sync_data,
640
headers={'Authorization': f'Bearer {get_api_token()}'},
641
timeout=30
642
)
643
644
if response.status_code == 200:
645
# Update sync status
646
item.last_sync = datetime.datetime.now()
647
item.sync_status = 'SUCCESS'
648
self.datamodel.edit(item)
649
success_count += 1
650
651
else:
652
item.sync_status = 'FAILED'
653
self.datamodel.edit(item)
654
error_count += 1
655
656
except requests.RequestException as e:
657
logger.error(f"External sync failed for item {item.id}: {e}")
658
item.sync_status = 'ERROR'
659
self.datamodel.edit(item)
660
error_count += 1
661
662
message = f"Sync completed: {success_count} success, {error_count} failed"
663
flash_type = "success" if error_count == 0 else "warning"
664
flash(message, flash_type)
665
666
return redirect(self.get_redirect())
667
668
# Action with progress tracking
669
@action("batch_process", "Batch Process", "Start batch processing?", "fa-play")
670
def batch_process_with_progress(self, items):
671
"""Long-running batch process with progress tracking."""
672
if not items:
673
flash("No items selected", "warning")
674
return redirect(self.get_redirect())
675
676
# Create background task
677
task_id = str(uuid.uuid4())
678
679
# Store task info in session or cache
680
session[f'task_{task_id}'] = {
681
'total_items': len(items),
682
'processed_items': 0,
683
'status': 'RUNNING',
684
'started_at': datetime.datetime.now().isoformat()
685
}
686
687
# Start background processing
688
process_items_async.delay(task_id, [item.id for item in items])
689
690
flash(f"Batch processing started. Task ID: {task_id}", "info")
691
692
# Redirect to progress page
693
return redirect(url_for('PersonModelView.batch_progress', task_id=task_id))
694
695
@expose('/batch-progress/<task_id>/')
696
def batch_progress(self, task_id):
697
"""Show batch processing progress."""
698
task_info = session.get(f'task_{task_id}')
699
700
if not task_info:
701
flash("Task not found", "error")
702
return redirect(self.get_redirect())
703
704
return self.render_template(
705
'batch_progress.html',
706
task_id=task_id,
707
task_info=task_info
708
)
709
710
# Celery task for background processing
711
from celery import Celery
712
713
@celery.task
714
def process_items_async(task_id, item_ids):
715
"""Background task for processing items."""
716
from flask import current_app
717
718
with current_app.app_context():
719
for i, item_id in enumerate(item_ids):
720
# Process individual item
721
item = db.session.query(Person).get(item_id)
722
process_single_item(item)
723
724
# Update progress
725
update_task_progress(task_id, i + 1, len(item_ids))
726
727
# Mark task as completed
728
mark_task_completed(task_id)
729
```