Database-First CQRS

What is Database-First CQRS?

FraiseQL implements Command Query Responsibility Segregation (CQRS) at the database level. Queries execute against PostgreSQL Views (v_*, tv_*) optimized for reads. Mutations execute PostgreSQL Functions (fn_*) that enforce business logic with ACID guarantees. Your database becomes your domain model.

Pattern	PostgreSQL	GraphQL	Benefit
Query (Read)	Views: v_, tv_	✅ GraphQL Query	Optimized reads, denormalized
Mutation (Write)	Functions: fn_*	✅ GraphQL Mutation	ACID guarantees, validation
Business Logic	PostgreSQL Functions	✅ MutationResult type	Transactional, no eventual consistency
Error Handling	Structured returns	✅ Status, message, data	Type-safe error handling

How It Works

1. Define PostgreSQL Views (Queries)

-- v_users: Regular view for real-time data
CREATE VIEW v_users AS
SELECT
    id,
    email,
    data->>'name' as name,
    data->>'role' as role,
    created_at,
    organization_id
FROM users
WHERE deleted_at IS NULL;

-- tv_users_with_stats: Materialized view for performance
CREATE MATERIALIZED VIEW tv_users_with_stats AS
SELECT
    u.id,
    u.email,
    u.data->>'name' as name,
    COUNT(p.id) as post_count,
    MAX(p.created_at) as last_post_at
FROM users u
LEFT JOIN posts p ON p.user_id = u.id
WHERE u.deleted_at IS NULL
GROUP BY u.id, u.email, u.data;

-- Refresh materialized view periodically
CREATE INDEX idx_tv_users_stats ON tv_users_with_stats(id);

2. Define PostgreSQL Functions (Mutations)

-- fn_create_user: Mutation function with validation
CREATE OR REPLACE FUNCTION fn_create_user(
    p_email TEXT,
    p_name TEXT,
    p_organization_id INT
) RETURNS JSONB AS $$
DECLARE
    v_user_id INT;
    v_result JSONB;
BEGIN
    -- Validation
    IF p_email !~ '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z]{2,}$' THEN
        RETURN jsonb_build_object(
            'status', 'error',
            'message', 'Invalid email format',
            'object_data', NULL
        );
    END IF;

    -- Check for duplicate
    IF EXISTS (SELECT 1 FROM users WHERE email = p_email) THEN
        RETURN jsonb_build_object(
            'status', 'error',
            'message', 'Email already exists',
            'object_data', NULL
        );
    END IF;

    -- Insert user
    INSERT INTO users (email, data, organization_id)
    VALUES (
        p_email,
        jsonb_build_object('name', p_name),
        p_organization_id
    )
    RETURNING id INTO v_user_id;

    -- Audit log (automatic)
    INSERT INTO tb_entity_change_log (
        entity_type, entity_id, operation, changed_by
    ) VALUES (
        'user', v_user_id, 'create', current_user
    );

    -- Success response
    RETURN jsonb_build_object(
        'status', 'success',
        'message', 'User created successfully',
        'object_data', jsonb_build_object(
            'id', v_user_id,
            'email', p_email,
            'name', p_name
        )
    );
END;
$$ LANGUAGE plpgsql;

3. Map to GraphQL

from fraiseql import fraiseql
from fraiseql.mutations import MutationResult

# Query: Maps to v_users view
@fraiseql.type
class User:
    id: int
    email: str
    name: str
    role: str
    post_count: int  # From tv_users_with_stats

@fraiseql.query
async def users(info) -> list[User]:
    """Executes: SELECT * FROM v_users"""
    return await info.context.db.fetch_all()

@fraiseql.query
async def users_with_stats(info) -> list[User]:
    """Executes: SELECT * FROM tv_users_with_stats"""
    return await info.context.db.fetch_all()

# Mutation: Calls fn_create_user function
@fraiseql.mutation
async def create_user(
    info,
    email: str,
    name: str,
    organization_id: int
) -> MutationResult:
    """Executes: SELECT fn_create_user($1, $2, $3)"""
    result = await info.context.db.call_function(
        "fn_create_user",
        email, name, organization_id
    )
    return MutationResult(**result)

Key Benefits

🔒

ACID Guarantees

Business logic in PostgreSQL functions = transactional consistency. No eventual consistency, no race conditions.

⚡

Optimized Reads

Views denormalize data for fast queries. Materialized views (tv_*) pre-compute expensive aggregations.

🛡️

Enforced Validation

Database-level constraints and validation. Cannot bypass via direct SQL or other clients.

📊

Structured Error Handling

MutationResult type with status, message, data. Type-safe error handling in GraphQL.

🏗️

True Separation

Queries can't modify data. Mutations enforce business rules. CQRS at database level.

📝

Automatic Audit Trails

Functions log to tb_entity_change_log automatically. Enterprise compliance built-in.

MutationResult Type

FraiseQL provides a standardized mutation response pattern:

from fraiseql.mutations import MutationResult

# GraphQL type automatically generated:
type MutationResult {
  status: String!      # "success", "error", "warning"
  message: String!     # Human-readable message
  object_data: JSON    # The created/updated object
}

# Example mutation response:
{
  "status": "success",
  "message": "User created successfully",
  "object_data": {
    "id": 123,
    "email": "user@example.com",
    "name": "Jane Doe"
  }
}

# Example error response:
{
  "status": "error",
  "message": "Email already exists",
  "object_data": null
}

Approach	Consistency	Performance	Complexity
Application-Level CQRS	❌ Eventual consistency	⚠️ Requires event bus	❌ High (Kafka, events, etc.)
Traditional ORM	✅ ACID	❌ N+1 queries	⚠️ Medium (DataLoaders)
GraphQL Resolvers	⚠️ Depends on code	❌ Slow (multiple queries)	⚠️ Medium (manual optimization)
FraiseQL Database-First	✅ ACID guaranteed	✅ Single query, views	✅ Low (database-driven)

🏗️ Database-First CQRS