Advanced Patterns

Master advanced FraiseQL patterns for production applications.

Soft Delete Pattern

Implement undo/restore functionality without losing data.

Schema Definition

import fraiseql
from fraiseql.scalars import ID, DateTime

@fraiseql.type
class Post:
    id: ID
    identifier: str
    title: str
    content: str
    created_at: DateTime
    deleted_at: DateTime | None  # NULL = not deleted

@fraiseql.query
def posts() -> list[Post]:
    """Get all non-deleted posts. The v_post_active view filters deleted_at IS NULL."""
    return fraiseql.config(sql_source="v_post_active")

@fraiseql.query
def posts_including_deleted() -> list[Post]:
    """Get all posts including deleted ones (admin only)."""
    return fraiseql.config(sql_source="v_post")

@fraiseql.mutation(sql_source="fn_delete_post", operation="DELETE")
def delete_post(id: ID) -> Post:
    """Soft delete: fn_delete_post sets deleted_at = NOW()."""
    pass

@fraiseql.mutation(sql_source="fn_restore_post", operation="UPDATE")
def restore_post(id: ID) -> Post:
    """Restore deleted post: fn_restore_post sets deleted_at = NULL."""
    pass

Database Implementation

-- Base table (trinity pattern)
CREATE TABLE tb_post (
    pk_post    BIGINT  GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    id         UUID    DEFAULT gen_random_uuid() UNIQUE NOT NULL,
    identifier TEXT    UNIQUE NOT NULL,        -- slug
    fk_user BIGINT NOT NULL REFERENCES tb_user(pk_user),
    title      TEXT    NOT NULL,
    content    TEXT    NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    deleted_at TIMESTAMPTZ                    -- NULL = not deleted
);

CREATE UNIQUE INDEX idx_tb_post_id ON tb_post(id);
CREATE INDEX idx_tb_post_fk_user ON tb_post(fk_user);
CREATE INDEX idx_tb_post_deleted_at ON tb_post(deleted_at)
    WHERE deleted_at IS NULL;

-- Active posts view (filters soft-deleted rows)
CREATE VIEW v_post_active AS
SELECT
    p.id,
    jsonb_build_object(
        'id',         p.id::text,
        'identifier', p.identifier,
        'title',      p.title,
        'content',    p.content,
        'createdAt',  p.created_at
    ) AS data
FROM tb_post p
WHERE p.deleted_at IS NULL;

-- All posts view (includes deleted)
CREATE VIEW v_post AS
SELECT
    p.id,
    jsonb_build_object(
        'id',         p.id::text,
        'identifier', p.identifier,
        'title',      p.title,
        'content',    p.content,
        'createdAt',  p.created_at,
        'deletedAt',  p.deleted_at
    ) AS data
FROM tb_post p;

Soft Delete Function

CREATE OR REPLACE FUNCTION fn_delete_post(p_id UUID)
RETURNS mutation_response LANGUAGE plpgsql AS $$
DECLARE
    v_pk BIGINT;
    v_entity JSONB;
BEGIN
    UPDATE tb_post
    SET deleted_at = NOW()
    WHERE id = p_id AND deleted_at IS NULL
    RETURNING pk_post INTO v_pk;

    IF v_pk IS NULL THEN
        RETURN ROW('failed:not_found', 'Post not found or already deleted',
                   p_id, 'Post', NULL, NULL, NULL, NULL)::mutation_response;
    END IF;

    SELECT data INTO v_entity FROM v_post WHERE id = p_id;

    RETURN ROW('success', NULL, p_id, 'Post',
               v_entity, NULL, NULL, NULL)::mutation_response;
END;
$$;

Cascading Soft Deletes

-- Trigger function to cascade soft-delete to comments
CREATE OR REPLACE FUNCTION cascade_soft_delete_post_comments()
RETURNS TRIGGER AS $$
BEGIN
    UPDATE tb_comment
    SET deleted_at = NOW()
    WHERE fk_post = NEW.pk_post AND deleted_at IS NULL;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- Only fires when deleted_at transitions from NULL to a value
CREATE TRIGGER cascade_post_delete
AFTER UPDATE ON tb_post
FOR EACH ROW
WHEN (NEW.deleted_at IS NOT NULL AND OLD.deleted_at IS NULL)
EXECUTE FUNCTION cascade_soft_delete_post_comments();

Audit Trail Pattern

Track all changes for compliance and debugging.

Schema Definition

from enum import Enum
import fraiseql
from fraiseql.scalars import ID, DateTime

@fraiseql.enum
class AuditAction(Enum):
    CREATE = "CREATE"
    UPDATE = "UPDATE"
    DELETE = "DELETE"
    RESTORE = "RESTORE"

@fraiseql.type
class AuditLog:
    id: ID
    entity_type: str
    entity_id: ID         # UUID of the affected entity (exposed in GraphQL)
    action: AuditAction
    old_values: dict | None
    new_values: dict | None
    changed_by: ID        # UUID of the acting user
    changed_at: DateTime

@fraiseql.query
def audit_logs(entity_type: str, entity_id: ID) -> list[AuditLog]:
    """Get audit trail for entity."""
    return fraiseql.config(sql_source="v_audit_log")

Database Implementation

-- Audit log base table (trinity pattern)
CREATE TABLE tb_audit_log (
    pk_audit_log BIGINT  GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    id           UUID    DEFAULT gen_random_uuid() UNIQUE NOT NULL,
    identifier   TEXT    UNIQUE NOT NULL,   -- auto-generated reference (e.g. uuid)
    entity_type  TEXT    NOT NULL,
    entity_id    UUID    NOT NULL,          -- UUID of the affected entity
    fk_user BIGINT REFERENCES tb_user(pk_user),  -- who made the change
    action       TEXT    NOT NULL,
    old_values   JSONB,
    new_values   JSONB,
    changed_at   TIMESTAMPTZ DEFAULT NOW()
);

CREATE UNIQUE INDEX idx_tb_audit_log_id ON tb_audit_log(id);
CREATE INDEX idx_tb_audit_log_entity ON tb_audit_log(entity_type, entity_id);
CREATE INDEX idx_tb_audit_log_fk_user ON tb_audit_log(fk_user);
CREATE INDEX idx_tb_audit_log_changed_at ON tb_audit_log(changed_at DESC);

-- Audit log read view
CREATE VIEW v_audit_log AS
SELECT
    a.id,
    jsonb_build_object(
        'id',          a.id::text,
        'entityType',  a.entity_type,
        'entityId',    a.entity_id::text,
        'action',      a.action,
        'oldValues',   a.old_values,
        'newValues',   a.new_values,
        'changedAt',   a.changed_at
    ) AS data
FROM tb_audit_log a;

Auto-Trigger Audit Logging

-- Trigger function to log post changes
CREATE OR REPLACE FUNCTION log_post_changes()
RETURNS TRIGGER AS $$
DECLARE
    v_user_id UUID;
    v_action  TEXT;
BEGIN
    -- Read acting user from session variable (set by mutation function)
    v_user_id := current_setting('app.user_id', true)::UUID;
    v_action  := TG_OP;   -- 'INSERT', 'UPDATE', or 'DELETE'

    INSERT INTO tb_audit_log (identifier, entity_type, entity_id, action, old_values, new_values)
    SELECT
        gen_random_uuid()::text,
        'Post',
        COALESCE(NEW.id, OLD.id),
        v_action,
        CASE WHEN TG_OP = 'INSERT' THEN NULL ELSE row_to_json(OLD)::jsonb END,
        CASE WHEN TG_OP = 'DELETE' THEN NULL ELSE row_to_json(NEW)::jsonb END;

    RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql;

-- Trigger on tb_post
CREATE TRIGGER tb_post_audit_trigger
AFTER INSERT OR UPDATE OR DELETE ON tb_post
FOR EACH ROW
EXECUTE FUNCTION log_post_changes();

Pass the user ID from the JWT into your mutation functions using inject=, then set the PostgreSQL session variable inside the SQL function:

@fraiseql.mutation(
    sql_source="fn_update_post",
    operation="UPDATE",
    inject={"user_id": "jwt:sub"}  # JWT 'sub' claim is passed as SQL param
)
def update_post(id: ID, title: str, content: str) -> Post:
    """Update post. fn_update_post sets app.user_id for audit triggers."""
    pass

CREATE OR REPLACE FUNCTION fn_update_post(
    p_id      UUID,
    p_title   TEXT,
    p_content TEXT,
    p_user_id UUID   -- injected from JWT
)
RETURNS mutation_response LANGUAGE plpgsql AS $$
BEGIN
    -- Make the user UUID available to triggers via session variable
    PERFORM set_config('app.user_id', p_user_id::text, true);

    UPDATE tb_post
    SET title = p_title, content = p_content
    WHERE id = p_id;

    -- ... return mutation_response
END;
$$;

Versioning Pattern

Track content versions and allow rollback.

Schema Definition

import fraiseql
from fraiseql.scalars import ID, DateTime

@fraiseql.type
class PostVersion:
    id: ID
    identifier: str
    version_number: int
    title: str
    content: str
    created_at: DateTime

@fraiseql.type
class Post:
    id: ID
    identifier: str
    title: str
    content: str
    current_version: int
    created_at: DateTime
    updated_at: DateTime

@fraiseql.query
def post_versions(post_id: ID) -> list[PostVersion]:
    """Get all versions of a post."""
    return fraiseql.config(sql_source="v_post_version")

@fraiseql.mutation(sql_source="fn_update_post_versioned", operation="UPDATE")
def update_post(id: ID, title: str, content: str) -> Post:
    """Update post (creates new version in tb_post_version)."""
    pass

@fraiseql.mutation(sql_source="fn_revert_post_version", operation="UPDATE")
def revert_to_version(post_id: ID, version_number: int) -> Post:
    """Revert to specific version."""
    pass

Database Schema

-- Post base table (trinity pattern)
CREATE TABLE tb_post (
    pk_post          BIGINT  GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    id               UUID    DEFAULT gen_random_uuid() UNIQUE NOT NULL,
    identifier       TEXT    UNIQUE NOT NULL,
    fk_user BIGINT NOT NULL REFERENCES tb_user(pk_user),
    title            TEXT    NOT NULL,
    content          TEXT    NOT NULL,
    current_version  INTEGER NOT NULL DEFAULT 1,
    created_at       TIMESTAMPTZ DEFAULT NOW(),
    updated_at       TIMESTAMPTZ DEFAULT NOW()
);

CREATE UNIQUE INDEX idx_tb_post_id ON tb_post(id);

-- Post version table (trinity pattern — versions are entities too)
CREATE TABLE tb_post_version (
    pk_post_version BIGINT  GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    id              UUID    DEFAULT gen_random_uuid() UNIQUE NOT NULL,
    identifier      TEXT    UNIQUE NOT NULL,  -- e.g. "{post_slug}-v{n}"
    fk_post BIGINT NOT NULL REFERENCES tb_post(pk_post) ON DELETE CASCADE,
    version_number  INTEGER NOT NULL,
    title           TEXT    NOT NULL,
    content         TEXT    NOT NULL,
    created_at      TIMESTAMPTZ DEFAULT NOW(),
    UNIQUE (fk_post, version_number)
);

CREATE UNIQUE INDEX idx_tb_post_version_id ON tb_post_version(id);
CREATE INDEX idx_tb_post_version_fk_post ON tb_post_version(fk_post);

-- Version read view
CREATE VIEW v_post_version AS
SELECT
    pv.id,
    jsonb_build_object(
        'id',            pv.id::text,
        'versionNumber', pv.version_number,
        'title',         pv.title,
        'content',       pv.content,
        'createdAt',     pv.created_at
    ) AS data
FROM tb_post_version pv;

Cursor-Based Pagination

Efficient pagination for large datasets using Relay connections.

Using the Built-In relay=True Option

FraiseQL has built-in Relay connection support. Add relay=True to your query and the runtime generates the Connection, Edge, and PageInfo types automatically:

import fraiseql
from fraiseql.scalars import ID

@fraiseql.type
class Post:
    id: ID
    identifier: str
    title: str

# relay=True wraps the result in a Relay Connection automatically.
# The view must include pk_post in the JSONB for cursor generation.
@fraiseql.query
def posts(first: int = 10, after: str | None = None) -> list[Post]:
    """Cursor-based pagination via Relay connection."""
    return fraiseql.config(sql_source="v_post_relay", relay=True)

-- Relay view: must include pk_post for cursor generation.
-- FraiseQL strips pk_post before sending to the client.
CREATE VIEW v_post_relay AS
SELECT
    p.id,
    jsonb_build_object(
        'pk_post',    p.pk_post,       -- required for cursor; stripped from response
        'id',         p.id::text,
        'identifier', p.identifier,
        'title',      p.title
    ) AS data
FROM tb_post p
ORDER BY p.pk_post;

Field-Level Access Control

Use fraiseql.field(requires_scope=...) to restrict individual fields based on JWT scopes. This is compile-time metadata — the Rust runtime enforces it without any Python code running.

from typing import Annotated
import fraiseql
from fraiseql.scalars import ID, Decimal

@fraiseql.type
class Employee:
    id: ID
    name: str
    department: str
    # Only HR or admins may see salary
    salary: Annotated[
        Decimal | None,
        fraiseql.field(requires_scope="hr:view_salary", on_deny="mask")
    ]
    # PII — reject the query entirely if requester lacks scope
    tax_id: Annotated[
        str | None,
        fraiseql.field(requires_scope="hr:view_pii", on_deny="reject")
    ]

on_deny="mask" returns null for users who lack the scope (query still succeeds). on_deny="reject" (the default) fails the entire query with a FORBIDDEN error.

Row-Level Security (Multi-Tenancy)

Use PostgreSQL RLS together with inject= to enforce tenant isolation. The inject= parameter passes JWT claims as SQL parameters — no Python runtime logic required.

import fraiseql
from fraiseql.scalars import ID

@fraiseql.type
class Post:
    id: ID
    identifier: str
    title: str
    tenant_id: str

# tenant_id is extracted from the JWT and passed to the view/RLS policy
@fraiseql.query(inject={"tenant_id": "jwt:tenant_id"})
def posts() -> list[Post]:
    """Get posts for the current tenant only (enforced by RLS)."""
    return fraiseql.config(sql_source="v_post")

-- The RLS policy reads tenant_id set by the Rust runtime via app.tenant_id
CREATE POLICY tb_post_tenant_isolation ON tb_post
USING (tenant_id = current_setting('app.tenant_id'))
WITH CHECK (tenant_id = current_setting('app.tenant_id'));

ALTER TABLE tb_post ENABLE ROW LEVEL SECURITY;

Caching Expensive Queries

Use cache_ttl_seconds in fraiseql.config() to cache results in the Rust runtime. No Python decorator or library needed — this is a compile-time instruction to the Rust cache layer.

import fraiseql
from fraiseql.scalars import ID

@fraiseql.type
class UserStats:
    id: ID
    post_count: int
    total_likes: int
    follower_count: int

# Cache this expensive aggregation for 1 hour per user_id argument
@fraiseql.query
def user_stats(user_id: ID) -> UserStats | None:
    """Cache expensive aggregation for one hour."""
    return fraiseql.config(sql_source="v_user_stats", cache_ttl_seconds=3600)

Error Handling in Mutation Functions

FraiseQL decorator function bodies are never executed — they are compile-time schema declarations only. Input validation and error handling belong in the PostgreSQL function. Mutation functions return a mutation_response composite type:

@fraiseql.mutation(sql_source="fn_create_post", operation="CREATE")
def create_post(title: str, content: str) -> Post:
    """Create a new post. Validation is enforced by fn_create_post."""
    pass

CREATE OR REPLACE FUNCTION fn_create_post(p_title TEXT, p_content TEXT)
RETURNS mutation_response LANGUAGE plpgsql AS $$
DECLARE
    v_pk   BIGINT;
    v_id   UUID;
    v_slug TEXT;
BEGIN
    IF length(p_title) < 3 THEN
        RETURN ROW(
            'failed:validation', 'Title too short (minimum 3 characters)',
            NULL, 'Post', NULL, NULL, NULL,
            jsonb_build_object('code', 'INVALID_INPUT', 'field', 'title')
        )::mutation_response;
    END IF;

    IF length(p_content) < 10 THEN
        RETURN ROW(
            'failed:validation', 'Content too short (minimum 10 characters)',
            NULL, 'Post', NULL, NULL, NULL,
            jsonb_build_object('code', 'INVALID_INPUT', 'field', 'content')
        )::mutation_response;
    END IF;

    v_slug := lower(regexp_replace(p_title, '[^a-zA-Z0-9]+', '-', 'g'));

    INSERT INTO tb_post (identifier, title, content)
    VALUES (v_slug, p_title, p_content)
    RETURNING pk_post, id INTO v_pk, v_id;

    RETURN ROW(
        'success', NULL, v_id, 'Post',
        (SELECT data FROM v_post WHERE id = v_id),
        NULL, NULL, NULL
    )::mutation_response;
END;
$$;

The status field uses the convention 'success', 'failed:{reason}', or 'conflict:{reason}'. FraiseQL maps these to appropriate GraphQL error responses.

Testing Advanced Patterns

Database tests are the best fit for advanced patterns. They run in transactions that roll back after each test, so every test starts clean:

import pytest
import psycopg


class TestSoftDelete:
    def test_soft_delete_hides_post(self, db):
        """Soft-deleted posts must not appear in v_post_active."""
        with db.cursor() as cur:
            cur.execute("""
                INSERT INTO tb_user (username, email, identifier)
                VALUES ('testuser', 'test@example.com', 'test@example.com')
                RETURNING pk_user
            """)
            pk_user = cur.fetchone()[0]

            cur.execute("""
                INSERT INTO tb_post (fk_user, title, content, identifier)
                VALUES (%s, 'My Post', 'Some content here', 'my-post')
                RETURNING id
            """, (pk_user,))
            post_id = cur.fetchone()[0]

            # Soft-delete the post
            cur.execute("SELECT status FROM fn_delete_post(%s::uuid)", (post_id,))
            status = cur.fetchone()[0]
            assert status == "success"

            # Active view must not return deleted post
            cur.execute("SELECT COUNT(*) FROM v_post_active WHERE id = %s", (post_id,))
            assert cur.fetchone()[0] == 0

            # Full view still returns it
            cur.execute("SELECT COUNT(*) FROM v_post WHERE id = %s", (post_id,))
            assert cur.fetchone()[0] == 1

    def test_audit_trigger_records_update(self, db):
        """Audit trigger must record UPDATE with old and new values."""
        with db.cursor() as cur:
            cur.execute("""
                INSERT INTO tb_user (username, email, identifier)
                VALUES ('auditor', 'auditor@example.com', 'auditor@example.com')
                RETURNING pk_user
            """)
            pk_user = cur.fetchone()[0]

            cur.execute("""
                INSERT INTO tb_post (fk_user, title, content, identifier)
                VALUES (%s, 'Original', 'Content here please', 'original')
                RETURNING pk_post
            """, (pk_user,))
            pk_post = cur.fetchone()[0]

            cur.execute("""
                UPDATE tb_post SET title = 'Updated' WHERE pk_post = %s
            """, (pk_post,))

            cur.execute("""
                SELECT action, old_values->>'title', new_values->>'title'
                FROM tb_audit_log
                WHERE entity_type = 'Post'
                ORDER BY changed_at DESC
                LIMIT 1
            """)
            action, old_title, new_title = cur.fetchone()

        assert action == "UPDATE"
        assert old_title == "Original"
        assert new_title == "Updated"

Summary Table

Pattern	Use Case	Key Mechanism
Soft Delete	Undo/restore	`deleted_at` column + filtered view
Audit Trail	Compliance	PostgreSQL trigger → `tb_audit_log`
Versioning	Content history	`tb_post_version` + `fk_post` FK
Cursor Pagination	Large datasets	`relay=True` + `pk_` cursor column
Field-Level Access	Data masking / RBAC	`fraiseql.field(requires_scope=...)`
Tenant Isolation	Multi-tenancy	PostgreSQL RLS + `inject=` from JWT
Caching	Expensive queries	`cache_ttl_seconds=` on `@fraiseql.query`

Advanced Patterns

Advanced Patterns

Soft Delete Pattern

Schema Definition

Database Implementation

Soft Delete Function

Cascading Soft Deletes

Audit Trail Pattern

Schema Definition

Database Implementation

Auto-Trigger Audit Logging

Versioning Pattern

Schema Definition

Database Schema

Cursor-Based Pagination

Using the Built-In relay=True Option

Field-Level Access Control

Row-Level Security (Multi-Tenancy)

Caching Expensive Queries

Error Handling in Mutation Functions

Testing Advanced Patterns

Summary Table

See Also