Decorators

FraiseQL uses decorators (Python) and decorator functions (TypeScript) to define GraphQL schema elements. This reference covers all available decorators, the GraphQL SDL they generate, and how to handle errors within decorated functions.

Note

This page documents Python decorators. TypeScript uses @type(), @input(), @query(), @mutation() from import { type, input, query, mutation } from 'fraiseql'. Go uses struct tags: fraiseql:"fieldname". See TypeScript SDK and Go SDK.

Decorator Execution Order

When a request arrives, FraiseQL processes it through a fixed pipeline:

Request
  → Middleware (in registration order)
  → Authentication (JWT/API key verification)
  → Authorization (role + scope checks)
  → Resolver (SQL view or function execution)
  → Response

When multiple decorators protect the same resolver, they execute in this order:

1. @authenticated — JWT validation (fails fast if not logged in)
2. @requires_scope — OAuth scope check
3. @role — Role-based access check
4. Resolver executes
5. @after_mutation or @observer — Post-execution side effects

Note: @before_mutation exceptions abort the mutation and roll back the database transaction.

This means:

• @middleware functions run before any authentication check — they can read or modify the raw request
• Authentication failures short-circuit before reaching authorization or resolvers
• @query/@mutation requires_scope checks run after the token is verified
• @before_mutation hooks run inside the resolver phase, after all auth checks pass
• @after_mutation hooks run after the SQL function completes successfully

@type

Defines a GraphQL object type.

Python

import fraiseql

@fraiseql.type
class User:
    """A user in the system."""
    id: ID
    name: str
    email: str
    created_at: DateTime

TypeScript

import { type } from 'fraiseql';

@type()
class User {
    /** A user in the system. */
    id!: ID;
    name!: string;
    email!: string;
    createdAt!: Date;
}

Generated GraphQL SDL:

type User {
  """A user in the system."""
  id: ID!
  name: String!
  email: String!
  createdAt: DateTime!
}

Options

Option	Type	Description
database	str	Database for federated types
dataplane	str	Data plane ("graphql" or "arrow")
implements	list[str]	Interfaces to implement
relay	bool	Implement the Node interface and enable keyset cursor generation

When relay=True, the type implements the Node interface, the global node(id: ID!) query is added to the schema, and fraiseql compile validates that pk_{entity} is present in the corresponding SQL view.

Python

@fraiseql.type(database="analytics", implements=["Node"])
class UserAnalytics:
    id: ID
    page_views: int

TypeScript

@type({ database: 'analytics', implements: ['Node'] })
class UserAnalytics {
    id!: ID;
    pageViews!: number;
}

Field Options

Use Annotated (Python) or field decorators (TypeScript) for field-level options:

Python

from typing import Annotated

@fraiseql.type
class User:
    id: ID
    email: str

    # Protected field
    salary: Annotated[Decimal, fraiseql.field(requires_scope="hr:read")]

    # Deprecated field
    username: Annotated[str, fraiseql.field(deprecated="Use email instead")]

    # Non-nullable with default
    role: Annotated[str, fraiseql.field(default="user")]

    # Computed field (not from database)
    display_name: Annotated[str, fraiseql.field(computed=True)]

TypeScript

import { type, field } from 'fraiseql';

@type()
class User {
    id!: ID;
    email!: string;

    @field({ requiresScope: 'hr:read' })
    salary!: number;

    @field({ deprecated: 'Use email instead' })
    username!: string;

    @field({ default: 'user' })
    role!: string;

    @field({ computed: true })
    displayName!: string;
}

@query

Defines a GraphQL query.

Python

@fraiseql.query(sql_source="v_user")
def user(id: ID) -> User | None:
    """Get a user by ID."""
    pass

TypeScript

import { query, ID } from 'fraiseql';

@query({ sqlSource: 'v_user' })
function user(id: ID): Promise<User | null> {
    return Promise.resolve(null);
}

Generated GraphQL SDL:

type Query {
  user(id: ID!): User
}

Options

Option	Type	Default	Description
sql_source	str	—	View or table to query
relay	bool	False	Enable Relay spec (Connection/Edge/PageInfo, keyset cursors)
auto_params	bool | dict	inferred	True = all four params; False = none; dict for partial overrides (e.g. {"limit": True, "offset": False}). Partial dicts merge with [query_defaults] in fraiseql.toml.
inject	dict[str, str]	—	Server-side JWT claim injection. Keys are SQL param names; values are "jwt:<claim>". See Server-Side Injection.
cache_ttl	int	—	Cache duration in seconds
cache	bool	—	Enable/disable caching
requires_role	str	—	Required role
requires_scope	str	—	Required scope

Auto Params

For list queries (-> list[T]), all four params are enabled automatically. Pass auto_params=False to opt out:

Param	GraphQL argument	SQL
where	where: {Type}WhereInput	WHERE …
orderBy	orderBy: [{Type}OrderByInput!]	ORDER BY …
limit	limit: Int	LIMIT …
offset	offset: Int	OFFSET …

# All four auto-enabled (list return type)
@fraiseql.query(sql_source="v_post")
def posts() -> list[Post]:
    pass

# Opt out entirely
@fraiseql.query(sql_source="v_post", auto_params=False)
def posts() -> list[Post]:
    pass

Relay mode

When relay=True, the query generates a Relay-spec {Type}Connection instead of [Type!]!, and limit/offset are replaced by first/after/last/before. Requires @fraiseql.type(relay=True) on the return type and pk_{entity} in the SQL view.

@fraiseql.type(relay=True)
class Post:
    id: UUID
    title: str

@fraiseql.query(sql_source="v_post", relay=True)
def posts() -> list[Post]:
    pass

See Relay Pagination for full details.

@mutation

Defines a GraphQL mutation.

Python

@fraiseql.mutation(sql_source="fn_create_user", operation="CREATE")
def create_user(email: str, name: str) -> User:
    """Create a new user."""
    pass

TypeScript

import { mutation } from 'fraiseql';

@mutation({ sqlSource: 'fn_create_user', operation: 'CREATE' })
function createUser(email: string, name: string): Promise<User> {
    return Promise.resolve({} as User);
}

Generated GraphQL SDL:

type Mutation {
  createUser(email: String!, name: String!): User!
}

Options

Option	Type	Description
sql_source	str	SQL function to call
operation	str	"CREATE", "UPDATE", "DELETE"
inject	dict[str, str]	Server-side JWT claim injection. Keys are SQL param names; values are "jwt:<claim>". Injected params are never exposed to clients. See Server-Side Injection.
requires_role	str	Required role
requires_scope	str	Required scope
cache_update	bool	Update cache after mutation

Python

@fraiseql.mutation(
    sql_source="fn_delete_user",
    operation="DELETE",
    requires_scope="admin:delete_user"
)
def delete_user(id: ID) -> bool:
    """Delete a user. Requires admin scope."""
    pass

TypeScript

@mutation({
    sqlSource: 'fn_delete_user',
    operation: 'DELETE',
    requiresScope: 'admin:delete_user',
})
function deleteUser(id: ID): Promise<boolean> {
    return Promise.resolve(true);
}

Error Handling

Raise errors from within @before_mutation hooks or @after_mutation hooks to abort or signal failures:

Python

# Raise errors from PostgreSQL functions using RAISE EXCEPTION —
# not from Python hooks. The Python layer is schema authoring only.
#
# In your PostgreSQL function:
# IF user_is_system THEN
#     RAISE EXCEPTION 'Cannot delete system users'
#         USING ERRCODE = 'P0001', HINT = 'FORBIDDEN';
# END IF;

TypeScript

import { beforeMutation, FraiseQLError } from 'fraiseql';

beforeMutation('deleteUser', async (args, context) => {
    const user = await getUser(args.id);
    if (!user) {
        throw new FraiseQLError('NOT_FOUND', `User ${args.id} does not exist`);
    }
    if (user.isSystemUser) {
        throw new FraiseQLError('FORBIDDEN', 'Cannot delete system users');
    }
});

FraiseQLError produces a structured GraphQL error with an extensions.code field:

{
  "errors": [
    {
      "message": "Cannot delete system users",
      "extensions": { "code": "FORBIDDEN" }
    }
  ]
}

@subscription

Defines a GraphQL subscription.

Python

@fraiseql.subscription(entity_type="User", topic="user_created")
def user_created() -> User:
    """Subscribe to new users."""
    pass

TypeScript

import { subscription } from 'fraiseql';

@subscription({ entityType: 'User', topic: 'user_created' })
function userCreated(): AsyncIterator<User> {
    return {} as AsyncIterator<User>;
}

Generated GraphQL SDL:

type Subscription {
  userCreated: User!
}

Options

Option	Type	Description
entity_type	str	Entity to watch
topic	str	NATS topic
operation	str	"INSERT", "UPDATE", "DELETE"
filter	str	Filter expression
jetstream	bool	Use JetStream
replay	bool	Allow replay

Python

@fraiseql.subscription(
    entity_type="Order",
    operation="UPDATE",
    filter="status == 'shipped'"
)
def order_shipped(customer_id: ID | None = None) -> Order:
    """Subscribe to shipped orders."""
    pass

TypeScript

@subscription({
    entityType: 'Order',
    operation: 'UPDATE',
    filter: "status == 'shipped'",
})
function orderShipped(customerId?: ID): AsyncIterator<Order> {
    return {} as AsyncIterator<Order>;
}

@input

Defines a GraphQL input type.

@fraiseql.input
class CreateUserInput:
    """Input for creating a user."""
    email: str
    name: str
    bio: str | None = None
    role: str = "user"

Generated GraphQL SDL:

input CreateUserInput {
  """Input for creating a user."""
  email: String!
  name: String!
  bio: String
  role: String!
}

Validation

from typing import Annotated

@fraiseql.input
class CreateUserInput:
    email: Annotated[str, fraiseql.validate(
        pattern=r"^[^@]+@[^@]+\.[^@]+$",
        message="Invalid email format"
    )]
    name: Annotated[str, fraiseql.validate(
        min_length=2,
        max_length=100
    )]
    age: Annotated[int, fraiseql.validate(
        minimum=0,
        maximum=150
    )]

@enum

Defines a GraphQL enum.

from enum import Enum

@fraiseql.enum
class OrderStatus(Enum):
    """Status of an order."""
    PENDING = "pending"
    CONFIRMED = "confirmed"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"

Usage:

@fraiseql.type
class Order:
    id: ID
    status: OrderStatus

@interface

Defines a GraphQL interface.

@fraiseql.interface
class Node:
    """An object with an ID."""
    id: ID

@fraiseql.interface
class Timestamped:
    """An object with timestamps."""
    created_at: DateTime
    updated_at: DateTime

Implement interfaces:

@fraiseql.type(implements=["Node", "Timestamped"])
class User:
    id: ID
    name: str
    created_at: DateTime
    updated_at: DateTime

@union

Defines a GraphQL union type.

@fraiseql.type
class User:
    id: ID
    name: str

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

@fraiseql.union(members=[User, Post])
class SearchResult:
    """Result from a search query."""
    pass

Usage:

@fraiseql.query(sql_source="fn_search")
def search(query: str) -> list[SearchResult]:
    pass

@scalar

Defines a custom scalar type.

@fraiseql.scalar
class Email:
    """Email address with validation."""

    @staticmethod
    def serialize(value: str) -> str:
        return value.lower()

    @staticmethod
    def parse(value: str) -> str:
        if "@" not in value:
            raise ValueError("Invalid email format")
        return value.lower()

    @staticmethod
    def parse_literal(ast) -> str:
        if isinstance(ast, StringValueNode):
            return Email.parse(ast.value)
        raise ValueError("Email must be a string")

@observer

Defines an event observer.

from fraiseql import observer, webhook, email, slack

@observer(
    entity="Order",
    event="INSERT",
    condition="total > 1000",
    actions=[
        webhook("https://api.example.com/orders"),
        slack("#sales", "New order: {id}")
    ]
)
def on_high_value_order():
    """Triggered for high-value orders."""
    pass

Options

Option	Type	Description
entity	str	Entity to watch
event	str	"INSERT", "UPDATE", "DELETE"
condition	str	Trigger condition
actions	list	Actions to execute
retry	RetryConfig	Retry configuration

Actions

# Webhook
webhook(
    url="https://api.example.com",
    headers={"Authorization": "Bearer {TOKEN}"},
    body_template='{"id": "{{id}}"}'
)

# Email
email(
    to="{customer_email}",
    subject="Order {id} confirmed",
    body="Your order has been confirmed."
)

# Slack
slack(
    channel="#orders",
    message="New order: {id} for ${total}"
)

@middleware

Defines request middleware.

@fraiseql.middleware
async def log_requests(request, next):
    """Log all requests."""
    start = time.time()
    response = await next(request)
    duration = time.time() - start
    logger.info(f"{request.operation} completed in {duration:.3f}s")
    return response

@after_mutation

Hook that runs after a mutation.

@fraiseql.after_mutation("create_user")
async def after_create_user(user: User, context):
    """Send welcome email after user creation."""
    await send_welcome_email(user.email, user.name)

@before_mutation

Hook that runs before a mutation.

@fraiseql.before_mutation("delete_user")
async def before_delete_user(id: ID, context):
    """Validate deletion is allowed."""
    user = await get_user(id)
    if user.is_system_user:
        raise ValueError("Cannot delete system user")

Typed Error Unions

# Available in v2.0.0
@fraiseql.error
class InsufficientFundsError:
    code: str
    required_amount: float
    available_amount: float

@fraiseql.mutation(sql_source="fn_transfer_funds")
def transfer_funds(
    from_id: str,
    to_id: str,
    amount: float
) -> Transfer | InsufficientFundsError:
    ...

This would generate a GraphQL union return type instead of raising a GraphQL error — allowing clients to handle domain errors as typed data rather than error extensions. Track progress in the FraiseQL Python SDK repository.

@observer and @after_mutation: Choosing the Right Hook

Both @observer and @after_mutation can react to data changes. They serve different purposes:

@after_mutation runs synchronously in the same transaction (blocks the response until it completes). @observer runs asynchronously after commit (fire-and-forget via NATS — does not block the response).

	@after_mutation	@observer
Trigger	Specific named mutation	Any insert/update/delete on an entity
Runs	In-process, synchronously after mutation	Asynchronously via NATS event
Use for	Tightly-coupled side effects (audit log, cache bust)	Loosely-coupled notifications (email, webhook, Slack)
Access to context	Full request context (user, headers)	Entity data only
Retries	No built-in retry	Configurable retry with backoff

Use @after_mutation when you need the full request context and want the side effect to be part of the same request lifecycle:

@fraiseql.after_mutation("create_user")
async def after_create_user(user: User, context):
    """Bust cache for user lists immediately after creation."""
    await context.cache.invalidate("users:*")

Use @observer when you need reliable delivery, retries, or want to decouple the side effect from the mutation:

@observer(
    entity="User",
    event="INSERT",
    actions=[
        email(to="{email}", subject="Welcome!", body="Hi {name}"),
    ],
    retry=RetryConfig(max_attempts=3, backoff_strategy="exponential"),
)
def on_user_created():
    """Send welcome email — retried on failure, decoupled from mutation."""
    pass

Tip

If both @after_mutation and @observer are registered for the same event, @after_mutation runs first (synchronously), then the observer fires asynchronously via NATS.

@arrow_query

Defines an Arrow Flight query.

@fraiseql.arrow_query(sql_source="va_analytics")
def analytics_data(
    start_date: Date,
    end_date: Date
) -> list[AnalyticsRow]:
    """Query analytics via Arrow Flight."""
    pass

fraiseql.field()

Field-level configuration.

fraiseql.field(
    requires_scope="scope",     # Required scope
    deprecated="message",       # Deprecation message
    default=value,              # Default value
    filterable=True,            # Include in WhereInput
    orderable=True,             # Include in OrderByInput
    computed=False,             # Is computed (not in DB)
    filter_type=CustomFilter,   # Custom filter type
    filter_operators=["_eq"],   # Allowed operators
)

fraiseql.validate()

Input validation rules.

fraiseql.validate(
    pattern="regex",          # Regex pattern
    min_length=1,             # Minimum string length
    max_length=100,           # Maximum string length
    minimum=0,                # Minimum number value
    maximum=100,              # Maximum number value
    enum=["a", "b"],          # Allowed values
    custom=validator_func,    # Custom validator
    message="Error message",  # Custom error message
)

Type Annotations

Optional Fields

@fraiseql.type
class User:
    bio: str | None  # Nullable field

Lists

@fraiseql.type
class User:
    roles: list[str]        # Non-null list, non-null items
    tags: list[str] | None  # Nullable list

Forward References

@fraiseql.type
class User:
    posts: list['Post']     # Forward reference
    manager: 'User | None'  # Self-reference

Export

Export schema to JSON:

## Quick Reference

### Decorator Cheat Sheet

| Decorator | Purpose | Key Options |
|-----------|---------|-------------|
| `@type` | Define object type | `database`, `implements` |
| `@input` | Define input type | — |
| `@query` | Define query | `sql_source`, `auto_params`, `cache_ttl` |
| `@mutation` | Define mutation | `sql_source`, `operation`, `requires_scope` |
| `@subscription` | Define subscription | `entity_type`, `filter`, `topic` |
| `@enum` | Define enum | — |
| `@interface` | Define interface | — |
| `@union` | Define union | `members` |
| `@scalar` | Define custom scalar | — |
| `@observer` | Event observer | `entity`, `event`, `condition`, `actions` |
| `@role` | Define role | — |
| `@middleware` | Request middleware | — |
| `@before_mutation` | Pre-mutation hook | — |
| `@after_mutation` | Post-mutation hook | — |
| `@arrow_query` | Arrow Flight query | `sql_source` |

### Common Auto Params

```python
@fraiseql.query(
    sql_source="v_user",
    auto_params={
        "limit": True,        # Add limit: Int parameter
        "offset": True,       # Add offset: Int parameter
        "where": True,        # Add where: UserWhereInput
        "order_by": True,     # Add orderBy: UserOrderByInput
        "total_count": True,  # Wrap result with totalCount
    }
)

Field Options

# With Annotated
from typing import Annotated

@fraiseql.type
class User:
    id: ID
    email: Annotated[str, fraiseql.field(
        requires_scope="user:read_email",  # Require scope
        deprecated="Use contactEmail instead",  # Deprecation
        default="",  # Default value
        computed=True,  # Computed field
    )]

Validation Rules

@fraiseql.input
class CreateUserInput:
    email: Annotated[str, fraiseql.validate(
        pattern=r"^[^@]+@[^@]+\.[^@]+$",
        message="Invalid email format"
    )]
    name: Annotated[str, fraiseql.validate(
        min_length=2,
        max_length=100
    )]
    age: Annotated[int, fraiseql.validate(
        minimum=13,
        maximum=120
    )]

Verify It Works

1. Define a simple type with decorators:

   import fraiseql
   
   @fraiseql.type
   class Product:
       id: str
       name: str
       price: float

2. Add a query decorator:

   @fraiseql.query(sql_source="v_product")
   def product(id: str) -> Product:
       pass

3. Compile the schema:

   fraiseql compile

   Expected output:

   ✓ Schema validated (1 types, 1 queries)
   ✓ Compiled to schema.compiled.json

4. Introspect the generated schema:

   curl -X POST http://localhost:8080/graphql \
     -H "Content-Type: application/json" \
     -d '{"query": "{ __type(name: \"Product\") { fields { name } } }"}' | jq

   Expected response:

   {
     "data": {
       "__type": {
         "fields": [
           {"name": "id"},
           {"name": "name"},
           {"name": "price"}
         ]
       }
     }
   }

5. Test the query:

   curl -X POST http://localhost:8080/graphql \
     -H "Content-Type: application/json" \
     -d '{"query": "{ product(id: \"123\") { id name price } }"}'

6. Test with authorization:

   @fraiseql.mutation(
       sql_source="fn_create_product",
       requires_scope="admin:products"
   )
   def create_product(name: str, price: float) -> Product:
       pass

   Test without scope (should fail):

   curl -X POST http://localhost:8080/graphql \
     -H "Authorization: Bearer $USER_TOKEN" \
     -d '{"query": "mutation { createProduct(name: \"Widget\", price: 9.99) { id } }"}'

   Expected error:

   {
     "errors": [{
       "message": "Forbidden: missing scope admin:products"
     }]
   }

Troubleshooting

Decorator Not Found

@fraiseql.type
class User:
    pass
# Error: module 'fraiseql' has no attribute 'type'

Solution: Use correct import:

import fraiseql

@fraiseql.type  # Correct
class User:
    pass

Type Not Generated

$ fraiseql compile
Warning: Type 'User' has no fields

Check:

1. All fields have type annotations
2. No syntax errors in class definition
3. Class is properly indented under decorator

Forward Reference Error

@fraiseql.type
class User:
    posts: list[Post]  # Error: Post not defined

Solution: Use string literal for forward reference:

@fraiseql.type
class User:
    posts: list['Post']  # Forward reference as string

Mutation Not Working

@fraiseql.mutation(sql_source="fn_create_user")
def create_user(name: str) -> User:
    pass

Check:

1. SQL function exists: SELECT * FROM pg_proc WHERE proname = 'fn_create_user'
2. Function returns correct type
3. Input parameters match function signature

Next Steps

Type System

Type System — Type definitions

Scalars

Scalars — Built-in scalars

GraphQL API

GraphQL API — Generated API