FraiseQL Extension Points: Customizing and Extending the Framework

Audience: Application developers, integration engineers, framework users

---

Executive Summary

FraiseQL v1 is a Python runtime GraphQL framework for PostgreSQL. You define your schema with decorators; at application startup the schema is built in memory and served over FastAPI. There is no compile step and no separate plugin runtime.

Extensibility is achieved through composition — you extend behavior by adding your own resolvers, scalars, middleware, and authorizers, and by pushing logic into PostgreSQL. The framework exposes the following real extension points:

Extension point	What it customizes
@fraiseql.field	Computed / derived fields on a type (sync or async)
@fraiseql.dataloader_field	Batched field resolution to prevent N+1 queries
Custom scalars	Domain-specific typed values
FastAPI / ASGI middleware	Cross-cutting request/response behavior
Authorizer + authorizer=	Per-operation and global authorization
PostgreSQL	The deepest surface: fn_ functions, triggers, custom types, extensions

---

1. Custom Field Resolvers

Use @fraiseql.field to add a computed or derived field to a @fraiseql.type. The decorated method becomes a GraphQL field whose value is produced at resolution time rather than read directly from the view's data JSONB.

import fraiseql


@fraiseql.field(resolver=None, description=None, track_n1=True)
def field(...): ...

The signature is:

• resolver — an optional custom resolver to override the default behavior (defaults to None, in which case the decorated method body is the resolver).
• description — the field description that appears in the GraphQL schema.
• track_n1 — whether to track N+1 query patterns for this field (default True).

1.1 Synchronous computed field

import fraiseql


@fraiseql.type(sql_source="v_user")
class User:
    id: fraiseql.ID
    first_name: str
    last_name: str

    @fraiseql.field(description="User's full display name")
    def display_name(self) -> str:
        return f"{self.first_name} {self.last_name}"

1.2 Async field with database access

The resolver receives the GraphQL info object, so it can reach the request-scoped repository through info.context["db"] and run additional reads.

import fraiseql
from uuid import UUID


@fraiseql.type(sql_source="v_user")
class User:
    id: UUID

    @fraiseql.field(description="Number of posts authored by this user")
    async def post_count(self, info) -> int:
        db = info.context["db"]
        return await db.fetchval(
            "SELECT count(*) FROM v_post WHERE author_id = $1", self.id
        )

Because the parent object (self) and info are both available, a custom field can combine already-loaded data with a targeted query. Keep these resolvers cheap: if a field issues one query per parent row, batch it with a dataloader instead (next section).

---

2. Dataloaders: Batching to Prevent N+1

When a field needs to look up a related record for every parent in a list, resolving it one row at a time produces an N+1 query pattern. @fraiseql.dataloader_field batches those lookups into a single load.

def dataloader_field(
    loader_class: type[DataLoader],
    *,
    key_field: str,
    description: str | None = None,
) -> ...: ...

• loader_class — a DataLoader subclass that knows how to batch-load by key.
• key_field — the attribute on the parent object holding the key to load.
• description — optional field description for the GraphQL schema.

The decorated method must have the signature (self, info) -> ReturnType; its body is auto-implemented by the decorator.

import fraiseql
from uuid import UUID
from fraiseql.optimization.dataloader import DataLoader


class UserDataLoader(DataLoader):
    async def batch_load(self, keys: list[UUID]) -> list["User | None"]:
        db = self.context["db"]
        rows = await db.find("v_user", id__in=keys)
        by_id = {row.id: row for row in rows}
        return [by_id.get(key) for key in keys]


@fraiseql.type(sql_source="v_post")
class Post:
    id: UUID
    author_id: UUID

    @fraiseql.dataloader_field(UserDataLoader, key_field="author_id")
    async def author(self, info) -> "User | None":
        """Load the post author. Batched across all posts in the request."""
        ...  # implementation generated by the decorator

All author lookups across a page of posts collapse into one batched load instead of one query per post.

See also: type system.

---

3. Custom Scalars

FraiseQL ships a library of domain scalars in fraiseql.types. Importing and annotating a field with one of these gives you parsing, serialization, and GraphQL schema typing for free.

import fraiseql
from fraiseql.types import ID, DateTime, EmailAddress, JSON, LTree


@fraiseql.type(sql_source="v_account")
class Account:
    id: ID
    email: EmailAddress      # validated email scalar
    created_at: DateTime     # ISO-8601 timestamp scalar
    metadata: JSON           # arbitrary JSON payload
    path: LTree              # PostgreSQL ltree hierarchical path

Available scalars include ID, Date, DateTime, EmailAddress, JSON, LTree, and many more domain types. Each scalar carries its own validation and serialization, so an EmailAddress field rejects malformed input at the GraphQL boundary before it ever reaches your resolver, and an LTree value maps cleanly onto PostgreSQL's ltree type.

For the complete catalog and the rules each scalar enforces, see the scalars reference.

---

4. FastAPI / ASGI Middleware

FraiseQL serves its GraphQL endpoint as a FastAPI application, so the entire FastAPI and Starlette middleware ecosystem is available to you. There are two ways to add cross-cutting request/response behavior.

4.1 Add middleware to the generated app

create_fraiseql_app returns a FastAPI instance. Add standard middleware to it with add_middleware:

import fraiseql
from fraiseql.fastapi import create_fraiseql_app
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware

app = create_fraiseql_app(
    database_url="postgresql://localhost/mydb",
    types=[User],
    queries=[users, user],
    mutations=[create_user],
    production=True,
)

app.add_middleware(GZipMiddleware, minimum_size=1024)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_methods=["POST"],
    allow_headers=["authorization", "content-type"],
)

4.2 Mount FraiseQL inside a larger FastAPI app

To run FraiseQL alongside your own REST routes and middleware, build your FastAPI app first and pass it to create_fraiseql_app via the app= parameter. FraiseQL extends that app with its GraphQL endpoint rather than creating a new one:

from fastapi import FastAPI
from fraiseql.fastapi import create_fraiseql_app

parent = FastAPI(title="My Service")


@parent.get("/healthz")
async def healthz() -> dict[str, str]:
    return {"status": "ok"}


# Add your own middleware to the parent app, then hand it to FraiseQL.
app = create_fraiseql_app(
    app=parent,
    database_url="postgresql://localhost/mydb",
    types=[User],
    queries=[users, user],
    mutations=[create_user],
)

Because middleware runs at the ASGI layer, it wraps every GraphQL request uniformly — ideal for request IDs, structured logging, compression, CORS, and security headers.

---

5. Authorization

FraiseQL enforces authorization at the operation level through the Authorizer protocol. An authorizer decides whether a top-level query, mutation, or subscription may run for the current request.

5.1 The Authorizer protocol

from typing import Any
from fraiseql.security import AuthorizationDecision


class Authorizer:
    async def authorize_operation(
        self,
        *,
        context: dict[str, Any],
        operation_type,        # OperationType (query / mutation / subscription)
        operation_name: str,
        arguments: dict[str, Any],
    ) -> AuthorizationDecision | bool:
        ...

authorize_operation may be sync or async and may return either a plain bool (sugar for allow/deny) or an AuthorizationDecision carrying a stable code, message, and optional row filters. Enforcement is fail-closed: if no authorizer is configured the operation is allowed; if a configured authorizer raises an unexpected error the operation is denied and the raw error is never surfaced to the client.

5.2 Global and per-operation authorizers

Pass an authorizer to create_fraiseql_app to apply it to every operation, and override it on a single operation with @fraiseql.query(authorizer=...) or @fraiseql.subscription(authorizer=...). The per-operation authorizer wins over the global default.

import fraiseql
from fraiseql.fastapi import create_fraiseql_app


class TenantAuthorizer:
    async def authorize_operation(self, *, context, operation_type, operation_name, arguments):
        return context.get("tenant_id") is not None


@fraiseql.query(authorizer=AdminOnlyAuthorizer())
async def all_tenants(info) -> list[Tenant]:
    db = info.context["db"]
    return await db.find("v_tenant")


app = create_fraiseql_app(
    database_url="postgresql://localhost/mydb",
    types=[Tenant],
    queries=[all_tenants],
    authorizer=TenantAuthorizer(),   # global default
)

5.3 Decision caching

Repeated authorization checks for the same context and operation can be served from a decision cache instead of re-running the authorizer. Enable it by passing an AuthorizationCacheConfig to create_fraiseql_app:

from fraiseql.fastapi import create_fraiseql_app
from fraiseql.security import AuthorizationCacheConfig

app = create_fraiseql_app(
    database_url="postgresql://localhost/mydb",
    types=[Tenant],
    queries=[all_tenants],
    authorizer=TenantAuthorizer(),
    authorization_cache=AuthorizationCacheConfig(),
)

A fresh cache hit replays the prior decision without calling the authorizer. Clean returns (allow or deny) are cached; an authorizer that raises hits the fail-closed branch and is never cached, so a transient error can neither pin a deny nor leak an allow.

5.4 Enterprise RBAC

For role-based access control at scale, the optional fraiseql.enterprise.rbac module provides hierarchical roles, PostgreSQL-native permission caching with automatic invalidation, and supporting GraphQL directives and middleware. Initialize it during startup with setup_rbac_cache(db_pool) and layer it on top of the Authorizer surface described above.

For the full authorization model, see the authorization guide.

---

6. PostgreSQL: The Deepest Extension Surface

In FraiseQL, PostgreSQL is not just storage — it is where most extension actually happens. Reads are served from v_/tv_ views and writes run through fn_ functions, so you can extend behavior end to end without touching the framework.

6.1 Business logic in fn_ functions

Every mutation calls a PostgreSQL function that performs validation and the write, then returns JSONB describing success or failure. This is the primary place to put domain logic.

CREATE FUNCTION fn_create_user(input jsonb)
RETURNS jsonb
LANGUAGE plpgsql
AS $$
DECLARE
    new_id uuid;
BEGIN
    IF NOT (input->>'email') ~ '^[^@]+@[^@]+$' THEN
        RETURN jsonb_build_object('success', false, 'message', 'invalid email');
    END IF;

    INSERT INTO tb_user (id, name, email)
    VALUES (gen_random_uuid(), input->>'name', input->>'email')
    RETURNING id INTO new_id;

    RETURN jsonb_build_object(
        'success', true,
        'user', jsonb_build_object('id', new_id, 'name', input->>'name')
    );
END;
$$;

import fraiseql


@fraiseql.mutation
async def create_user(info, input: CreateUserInput) -> CreateUserSuccess | CreateUserError:
    db = info.context["db"]
    result = await db.execute_function(
        "fn_create_user", {"name": input.name, "email": input.email}
    )
    if not result.get("success"):
        return CreateUserError(message=result.get("message", "failed"))
    return CreateUserSuccess(user=User(**result["user"]))

6.2 Triggers and custom types

Use triggers to keep projection views (tv_) in sync, maintain audit trails, or populate denormalized columns. Use PostgreSQL CREATE TYPE (composite or enum types) and DOMAIN constraints to model domain values at the database level — the shape your views return flows straight into the GraphQL response.

6.3 PostgreSQL extensions

Because views are just SQL, any installed PostgreSQL extension is available to your read and write paths:

• pgvector — similarity search over embeddings inside a v_ view.
• pg_trgm — fuzzy text matching and trigram indexes.
• PostGIS — geospatial queries and indexing.
• ltree — hierarchical paths, surfaced through the LTree scalar.

CREATE EXTENSION IF NOT EXISTS vector;

CREATE VIEW v_document_search AS
SELECT
    d.id,
    jsonb_build_object(
        'id', d.id,
        'title', d.title,
        'similarity', 1 - (d.embedding <=> :query_embedding)
    ) AS data
FROM tb_document d
ORDER BY d.embedding <=> :query_embedding;

The resolver queries this view like any other read; the extension does the heavy lifting inside PostgreSQL.

---

7. Best Practices

DO:

• Keep custom field resolvers cheap; batch related lookups with a dataloader.
• Use async/await for any I/O in resolvers, authorizers, and middleware.
• Push validation and write logic into fn_ functions where it stays transactional.
• Use the built-in domain scalars before reaching for a custom type.
• Make authorizers fail-closed and free of side effects.

DON'T:

• Issue one query per row in a field resolver — that is the N+1 pattern dataloaders exist to prevent.
• Expose internal pk_/fk_ columns through a view's data JSONB.
• Bypass the authorizer by reaching past info.context.
• Log sensitive data (tokens, passwords, full email lists) from middleware or hooks.

---

Related Documentation

• Integration patterns
• Error handling model
• Type system
• Scalars reference
• Decorators reference
• Authorization