CRUD operations

In-depth tour of POST, PATCH and DELETE — what gets validated, what gets auto-filled, what raises which status.

POST — create

Pipeline:

1. Body parsed (and validated against create_schema if set).
2. Keys are split into regular fields and custom_* fields.
3. create_fields whitelist enforced — any unknown key returns 403.
4. Auto-fill from session: created_by_id, updated_by_id, owner_id are populated from self.user['id'] when those columns exist on the model.
5. Per-field validation against the model: empty strings on blank=False fields raise 403, None on null=False fields raise 403.
6. model.objects.acreate(**to_save) runs.
7. IntegrityError is caught — duplicate-entry messages are rewritten to "<value> already exist" (409).
8. custom_* keys are saved through obj._custom.aset(...).
9. Many-to-many fields are saved via save_through_model(obj, name, ids).
10. Detail is fetched via get_obj(obj.id) and returned.
11. list:<model> cache namespace is invalidated.

PATCH — update

Pipeline:

1. id is parsed from the URL.
2. Body parsed (and validated against update_schema if set).
3. update_fields whitelist enforced — unknown key → 403.
4. update_fields not defined → 500 ("Update fields not defined").
5. Row fetched by id — and when owner_field is set, also filtered through _ownership_filter() so cross-owner rows return 404.
6. For each field in body:

• m2m field → getattr(obj, field).aset(value)
• custom_ field → obj._custom.aset(name, value)
• Foreign key → key rewritten to <field>_id, dict values reduced to id
• Other → setattr(obj, key, value) and tracked in self.diff

7. model.objects.filter(pk=id, **owner).aupdate(**to_update) — single SQL UPDATE, owner-scoped when owner_field is set.
8. Detail re-fetched via get_obj(id) and returned.
9. list:<model> and detail:<model>:<id> namespaces invalidated.

Diff tracking

Every field write populates self.diff with {old, new} pairs:

self.diff = {
    'name': {'old': 'Acme', 'new': 'Acme Inc'},
    'active': {'old': True, 'new': False},
}

Use it from post_process to log audit trails:

async def post_process(self, response):
    if self.diff and self.method == 'patch':
        await audit_log(self.user['id'], self.model, self.id, self.diff)
    return response

DELETE

Pipeline:

1. id parsed from the URL — no id → 404.
2. _ownership_filter() is applied. With owner_field set, rows owned by other users return 404.
3. queryset.filter(pk=id, **owner).adelete().
4. Failures bubble as 400 with the exception class + message.
5. Cache namespaces invalidated.

Response: {"success": true, "id": <id>, "message": "Deleted"}.

Method-not-allowed

When the request method is not in allowed_methods, dispatch raises HTTPException(405, '<METHOD> not allowed') before any handler runs.

Hooks per method

0-mcp by Stamatios Stamou Jr — github.com/ssjunior/0-mcp