Model Hooks¶

Hooks let you run custom logic at specific points in a model's lifecycle — before or after creating, updating, or deleting objects.

Warning

Do not call .save() / .asave() / .delete() / .adelete() on the same object (self) inside a hook — the framework detects re-entry and raises AmsdalRecursionError ("Trying to save/delete an object that is already being saved"). Calling these methods on other model instances is fine.

Initialization Hooks¶

pre_init¶

Called before the model is initialized and Pydantic validation runs. Use it to set default values or transform input data.

The kwargs dict contains the constructor arguments — modify it to change what gets passed to the model.

from typing import Any

def pre_init(self, *, is_new_object: bool, kwargs: dict[str, Any]) -> None:
    if is_new_object:
        kwargs['name'] = kwargs.get('name', 'Default Name')

    # Set a custom object ID based on another field.
    # NOTE: kwargs['_object_id'] takes effect only for models WITHOUT __primary_key__.
    # Custom-PK models derive their object_id from the named PK fields — set those instead.
    if kwargs.get('custom_id_field'):
        kwargs['_object_id'] = kwargs['custom_id_field']

Note

The Pydantic object is not fully initialized at this point. Access fields through kwargs, not self.

Note

is_new_object reflects the construction state, not the PK shape. A freshly-constructed model with a pre-set custom PK (e.g. Post(id='abc')) has is_new_object=True — save() correctly takes the create path. The flag is driven by self._state.adding, which the QuerySet sets to False when loading instances from the database.

post_init¶

Called after initialization and validation. The model instance is fully constructed.

from typing import Any

def post_init(self, *, is_new_object: bool, kwargs: dict[str, Any]) -> None:
    if self.name.islower():
        msg = 'Name must not be entirely lowercase'
        raise ValueError(msg)

Lifecycle Hooks¶

All lifecycle hooks accept only self. Each has a sync and async variant:

Event	Sync	Async
Before create	`pre_create`	`apre_create`
After create	`post_create`	`apost_create`
Before update	`pre_update`	`apre_update`
After update	`post_update`	`apost_update`
Before delete	`pre_delete`	`apre_delete`
After delete	`post_delete`	`apost_delete`

pre_create / apre_create¶

Called before a new object is saved to the database for the first time.

Note

save(force_insert=True) / asave(force_insert=True) always takes the create path, so pre_create / post_create (or their async variants) fire — never pre_update / post_update.

SyncAsync

def pre_create(self) -> None:
    if not self.name:
        self.name = 'Default Name'

async def apre_create(self) -> None:
    if not self.name:
        self.name = 'Default Name'

post_create / apost_create¶

Called after a new object is saved. Use it for side effects like creating related objects or sending notifications.

SyncAsync

def post_create(self) -> None:
    PersonProfile(person=self).save()

async def apost_create(self) -> None:
    await PersonProfile(person=self).asave()

pre_update / apre_update¶

Called before an existing object is updated. Use refetch_from_db() to compare with the current database state.

SyncAsync

def pre_update(self) -> None:
    original = self.refetch_from_db()

    if original.name != self.name:
        msg = 'Name cannot be changed'
        raise ValueError(msg)

async def apre_update(self) -> None:
    original = await self.arefetch_from_db()

    if original.name != self.name:
        msg = 'Name cannot be changed'
        raise ValueError(msg)

post_update / apost_update¶

Called after an existing object is updated.

SyncAsync

def post_update(self) -> None:
    send_email(self.email, subject=f'{self.name}, your profile was updated')

async def apost_update(self) -> None:
    send_email(self.email, subject=f'{self.name}, your profile was updated')

pre_delete / apre_delete¶

Called before an object is deleted. Raise an exception to prevent deletion.

SyncAsync

def pre_delete(self) -> None:
    if self.account_balance < 0:
        msg = 'Cannot delete account with negative balance'
        raise ValueError(msg)

async def apre_delete(self) -> None:
    if self.account_balance < 0:
        msg = 'Cannot delete account with negative balance'
        raise ValueError(msg)

post_delete / apost_delete¶

Called after an object is deleted.

SyncAsync

def post_delete(self) -> None:
    send_email(self.email, subject=f'{self.name}, your profile was deleted')

async def apost_delete(self) -> None:
    send_email(self.email, subject=f'{self.name}, your profile was deleted')

Note

Bulk operations (bulk_create, bulk_update, bulk_delete) do not trigger hooks.

Bypassing hooks per-call¶

All four single-object lifecycle methods accept a skip_hooks=False keyword. Pass True to skip the corresponding pre/post hooks for that one call only:

person.save(skip_hooks=True)              # no pre_create/post_create or pre_update/post_update
await person.asave(skip_hooks=True)       # same, async
person.delete(skip_hooks=True)            # no pre_delete/post_delete
await person.adelete(skip_hooks=True)

Use sparingly — it's intended for internal automation paths (e.g. migrations, rollback) where the hooks would re-trigger logic you've already executed.

Inspecting instance state¶

Every model instance carries a ModelState lifecycle namespace at instance._state:

Field	Meaning
`_state.adding`	`True` for a freshly-constructed instance; flipped to `False` after the first successful create; rolled back to `True` if `post_create` raises.
`_state.is_from_lakehouse`	`True` if the instance was loaded via `using('lakehouse')` (or in lakehouse-only mode); `False` for state-loaded or in-memory instances.

The public properties instance.is_new_object and instance.is_from_lakehouse delegate to _state.

There is no pre_save umbrella hook. To branch on lifecycle state, inspect _state inside the appropriate hook:

class Article(Model):
    title: str
    author: User | None = None

    def pre_create(self) -> None:
        # _state.adding is True here (still in the create path)
        assert self._state.adding

    def post_create(self) -> None:
        # _state.adding has been flipped to False by this point
        if not self._state.is_from_lakehouse:
            send_event('article_created', self.object_id)

force_insert=True re-asserts _state.adding=True before routing through _create, so pre_create / post_create fire (not pre_update).