
Type Safety Validation
- 12 installs
- 213 repo stars
- Updated August 4, 2026
- yonatangross/orchestkit
Helps with ai & agent building tasks.
About
type-safety-validation is a Claude Code skill for ai & agent building. It helps developers move faster with AI-assisted coding.
- type-safety-validation
- AI & Agent Building
- AI-coding skill
Type Safety Validation by the numbers
- 12 all-time installs (skills.sh)
- Ranked #11,618 of 16,546 AI & Agent Building skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
npx skills add https://github.com/yonatangross/orchestkit --skill type-safety-validationAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 12 |
|---|---|
| repo stars | ★ 213 |
| Last updated | August 4, 2026 |
| Repository | yonatangross/orchestkit ↗ |
What it does
Helps with ai & agent building tasks.
Files
Type Safety & Validation
Overview
When to use this skill:
- Building type-safe APIs (REST, RPC, GraphQL)
- Validating user input and external data
- Ensuring database queries are type-safe
- Creating end-to-end typed full-stack applications
- Implementing strict validation rules
Core Stack Quick Reference
| Tool | Purpose | Key Pattern |
|---|---|---|
| Zod | Runtime validation | z.object({}).safeParse(data) |
| tRPC | Type-safe APIs | t.procedure.input(schema).query() |
| Prisma | Type-safe ORM | Auto-generated types from schema |
| TypeScript 5.7+ | Compile-time safety | satisfies, const params, decorators |
Zod Essentials
import { z } from 'zod'
// Define schema
const UserSchema = z.object({
id: z.string().uuid(),
email: z.string().email(),
age: z.number().int().positive().max(120),
role: z.enum(['admin', 'user', 'guest']),
createdAt: z.date().default(() => new Date())
})
// Infer TypeScript type
type User = z.infer<typeof UserSchema>
// Validate with error handling
const result = UserSchema.safeParse(data)
if (result.success) {
const user: User = result.data
} else {
console.error(result.error.issues)
}See: references/zod-patterns.md for transforms, refinements, discriminated unions, and recursive types.
tRPC Essentials
import { initTRPC } from '@trpc/server'
import { z } from 'zod'
const t = initTRPC.create()
export const appRouter = t.router({
getUser: t.procedure
.input(z.object({ id: z.string() }))
.query(async ({ input }) => {
return await db.user.findUnique({ where: { id: input.id } })
}),
createUser: t.procedure
.input(z.object({ email: z.string().email(), name: z.string() }))
.mutation(async ({ input }) => {
return await db.user.create({ data: input })
})
})
export type AppRouter = typeof appRouterSee: references/trpc-setup.md for middleware, authentication, React integration, and error handling.
Exhaustive Type Checking
// ALWAYS use assertNever for compile-time exhaustiveness
function assertNever(x: never): never {
throw new Error("Unexpected value: " + x)
}
type Status = 'pending' | 'running' | 'completed' | 'failed'
function getStatusColor(status: Status): string {
switch (status) {
case 'pending': return 'gray'
case 'running': return 'blue'
case 'completed': return 'green'
case 'failed': return 'red'
default: return assertNever(status) // Compile-time check!
}
}
// Exhaustive record mapping
const statusColors = {
pending: 'gray',
running: 'blue',
completed: 'green',
failed: 'red',
} as const satisfies Record<Status, string>See: references/typescript-advanced.md for handler objects, type guards, and anti-patterns.
Branded Types
TypeScript (with Zod):
const UserId = z.string().uuid().brand<'UserId'>()
const AnalysisId = z.string().uuid().brand<'AnalysisId'>()
type UserId = z.infer<typeof UserId>
type AnalysisId = z.infer<typeof AnalysisId>
function deleteAnalysis(id: AnalysisId): void { ... }
deleteAnalysis(userId) // Error: UserId not assignable to AnalysisIdPython (with NewType):
from typing import NewType
from uuid import UUID
AnalysisID = NewType("AnalysisID", UUID)
ArtifactID = NewType("ArtifactID", UUID)
def delete_analysis(id: AnalysisID) -> None: ...
delete_analysis(artifact_id) # Error with mypy/tySee: references/typescript-advanced.md for factory patterns and pure TypeScript branding.
Python Type Safety with Ty
from typing import cast
# Type-safe extraction from untyped dict
result = {"findings": {...}, "confidence_score": 0.85}
findings_to_save: dict[str, object] | None = (
cast("dict[str, object]", result.get("findings"))
if isinstance(result.get("findings"), dict) else None
)
confidence_to_save: float | None = (
float(result.get("confidence_score"))
if isinstance(result.get("confidence_score"), (int, float)) else None
)See: references/ty-type-checker-patterns.md for mixed numeric handling and nested dict extraction.
References
| Reference | Content |
|---|---|
references/zod-patterns.md | Schemas, transforms, refinements, unions, recursion, error handling |
references/trpc-setup.md | Server setup, middleware, routers, client integration, subscriptions |
references/typescript-5-features.md | TS 5.0-5.7 features, satisfies, decorators, strict config |
references/typescript-advanced.md | Exhaustive patterns, branded types, type guards |
references/ty-type-checker-patterns.md | Python ty compliance, dict extraction, type narrowing |
references/prisma-types.md | Prisma ORM types, queries, relations |
Best Practices
Validation
- Validate at boundaries (API inputs, form submissions, external data)
- Use
.safeParse()to handle errors gracefully - Use branded types for IDs (
z.string().brand<'UserId'>())
Type Safety
- Enable
strict: trueintsconfig.json - Use
noUncheckedIndexedAccessfor safer array access - Prefer
unknownoverany - Exhaustive switches: Always use
assertNeverin default case - Exhaustive records: Use
satisfies Record<UnionType, Value>
Performance
- Reuse schemas (don't create inline in hot paths)
- Use
.parse()for known-good data (faster than.safeParse()) - Use tRPC batching for multiple queries
Resources
Related Skills
input-validation- Security-focused validation and sanitization patternsapi-design-framework- REST API design with type-safe contractsfastapi-advanced- Python backend with Pydantic type validation
Key Decisions
| Decision | Choice | Rationale |
|---|---|---|
| Runtime Validation | Zod | Best DX, excellent TypeScript inference, composable schemas |
| API Layer | tRPC | End-to-end type safety without code generation |
| Exhaustive Checks | assertNever | Compile-time guarantee for union completeness |
| Branded Types | Zod .brand() | Prevents ID type confusion with minimal overhead |
---
Skill Version: 1.2.0 Last Updated: 2025-12-27 Maintained by: AI Agent Hub Team
Capability Details
zod-schemas
Keywords: zod, schema, validation, parse, safeParse, infer, refine, transform Solves:
- How do I validate input with Zod?
- Create runtime validation schema
- Infer TypeScript types from Zod
- Transform and refine data with Zod
exhaustive-types
Keywords: exhaustive, assertNever, never assertion, switch exhaustive, compile-time exhaustiveness Solves:
- How do I make switch statements exhaustive?
- Compile-time check for missing union cases
- assertNever pattern for TypeScript
branded-types
Keywords: branded type, type branding, nominal type, NewType, brand, distinct types, id types Solves:
- How do I prevent mixing different ID types?
- Branded types with Zod
- Python NewType for type safety
trpc
Keywords: trpc, type-safe api, procedure, router, mutation, query, middleware Solves:
- How do I set up tRPC?
- Type-safe API calls
- tRPC with React Query
prisma-types
Keywords: prisma, orm, generated types, model, client, payload Solves:
- How do I use Prisma types?
- Type-safe database queries
typescript-5-features
Keywords: typescript 5, const parameters, satisfies, decorators, template literals Solves:
- Use TypeScript 5.x features
- Const type parameters
- Satisfies operator
ty-type-checker
Keywords: ty, rust type checker, strict typing, isinstance, cast, type narrowing Solves:
- How do I make ty type checker pass?
- Extract values from untyped dicts safely
- Type narrowing with isinstance checks
Type Safety Checklist
Comprehensive checklist for implementing end-to-end type safety in TypeScript applications.
Schema Definition
Zod Schema Design
- [ ] Define schemas for all API request/response types
- [ ] Create reusable base schemas (email, UUID, URL, etc.)
- [ ] Use
.inferto generate TypeScript types from schemas - [ ] Organize schemas by domain (user, post, comment, etc.)
- [ ] Export both schemas and inferred types
- [ ] Use
satisfiesoperator to ensure type compatibility - [ ] Document complex schemas with JSDoc comments
- [ ] Create discriminated unions for polymorphic types
- [ ] Use lazy schemas for recursive types
- [ ] Define custom error messages for better UX
TypeScript Configuration
- [ ] Enable
strict: truein tsconfig.json - [ ] Enable
strictNullChecks - [ ] Enable
strictFunctionTypes - [ ] Enable
noImplicitAny - [ ] Enable
noUncheckedIndexedAccess - [ ] Enable
noImplicitReturns - [ ] Enable
noFallthroughCasesInSwitch - [ ] Enable
exactOptionalPropertyTypes(TS 5.0+) - [ ] Set
moduleResolutionto "bundler" or "nodenext" - [ ] Enable
isolatedModulesfor build tools
Database Schema (Prisma)
- [ ] Define all models in schema.prisma
- [ ] Add appropriate indexes for query performance
- [ ] Use relations for foreign keys
- [ ] Add
@@indexdirectives for common queries - [ ] Use
@uniqueconstraints where appropriate - [ ] Add
@defaultvalues for fields - [ ] Use
@updatedAtfor automatic timestamp updates - [ ] Document models with triple-slash comments
- [ ] Run
prisma generateafter schema changes - [ ] Run
prisma migrate devto sync database
Validation Implementation
Input Validation
- [ ] Validate all user inputs with Zod
- [ ] Validate environment variables at startup
- [ ] Validate API request bodies
- [ ] Validate query parameters and path params
- [ ] Validate file uploads (size, type, name)
- [ ] Validate form submissions before API calls
- [ ] Use
.safeParse()to handle errors gracefully - [ ] Format Zod errors for user-friendly messages
- [ ] Validate data from external APIs
- [ ] Validate data from database before use
API Validation (tRPC)
- [ ] Define input schemas for all procedures
- [ ] Use Zod for runtime validation in
.input() - [ ] Return proper error codes (UNAUTHORIZED, NOT_FOUND, etc.)
- [ ] Implement authentication middleware
- [ ] Implement rate limiting middleware
- [ ] Implement logging middleware
- [ ] Use typed context for shared data
- [ ] Export
AppRoutertype for client - [ ] Configure error formatting for Zod errors
- [ ] Use superjson for Date/Map/Set serialization
Form Validation (React)
- [ ] Use react-hook-form with zodResolver
- [ ] Display validation errors inline
- [ ] Disable submit button during validation
- [ ] Show loading state during submission
- [ ] Handle API errors and display to user
- [ ] Implement optimistic updates where appropriate
- [ ] Reset form after successful submission
- [ ] Validate on blur for better UX
- [ ] Show field-level error messages
- [ ] Provide helpful error messages
Type Generation
OpenAPI/Swagger
- [ ] Generate OpenAPI spec from backend
- [ ] Use openapi-typescript to generate types
- [ ] Use openapi-zod-client to generate Zod schemas
- [ ] Automate type generation in CI/CD
- [ ] Version control generated types
- [ ] Document API endpoints in OpenAPI spec
- [ ] Include examples in OpenAPI spec
- [ ] Add security schemes to OpenAPI
- [ ] Generate client SDKs from OpenAPI
- [ ] Keep OpenAPI spec in sync with code
Prisma Type Generation
- [ ] Run
prisma generatein CI/CD - [ ] Use
Prisma.validatorfor reusable queries - [ ] Use
Prisma.UserGetPayloadto extract types - [ ] Create repository pattern for database access
- [ ] Type database transactions properly
- [ ] Use Prisma's generated types in API responses
- [ ] Extend Prisma types with computed fields
- [ ] Use
Omit/Pickto create DTOs from models - [ ] Generate Zod schemas from Prisma (zod-prisma-types)
- [ ] Keep Prisma schema as single source of truth
Type Sharing (Monorepo)
- [ ] Define shared types in common package
- [ ] Export API contract types from backend
- [ ] Import contract types in frontend
- [ ] Use path aliases for cleaner imports
- [ ] Version shared types package
- [ ] Document breaking changes
- [ ] Use TypeScript project references
- [ ] Keep shared types package lightweight
- [ ] Avoid circular dependencies
- [ ] Test shared types in isolation
Testing Type Safety
Unit Tests
- [ ] Test Zod schemas with valid inputs
- [ ] Test Zod schemas with invalid inputs
- [ ] Test error message formatting
- [ ] Test transformations and refinements
- [ ] Test async validators
- [ ] Test discriminated unions
- [ ] Test recursive schemas
- [ ] Mock database responses with correct types
- [ ] Test API error handling
- [ ] Achieve >80% coverage on validators
Integration Tests
- [ ] Test API endpoints with real HTTP calls
- [ ] Test tRPC procedures end-to-end
- [ ] Test database queries with real data
- [ ] Test authentication/authorization flows
- [ ] Test rate limiting
- [ ] Test error responses
- [ ] Test pagination (cursor and offset)
- [ ] Test file uploads
- [ ] Test SSE/WebSocket events
- [ ] Test transactions and rollbacks
Type Tests
- [ ] Use
expectTypeOffrom vitest for type tests - [ ] Test that inferred types match expected types
- [ ] Test that invalid types cause compile errors
- [ ] Test generic type parameters
- [ ] Test discriminated union exhaustiveness
- [ ] Test conditional types
- [ ] Test mapped types
- [ ] Use
@ts-expect-errorfor negative tests - [ ] Test branded types
- [ ] Test template literal types
Code Quality
Type Safety Best Practices
- [ ] Prefer
unknownoverany - [ ] Use type guards for narrowing
- [ ] Use branded types for domain primitives
- [ ] Use const assertions for literal types
- [ ] Use
satisfiesto preserve literal types - [ ] Avoid type assertions (
as) when possible - [ ] Use discriminated unions over plain unions
- [ ] Use exhaustive switch statements
- [ ] Leverage type inference (avoid redundant annotations)
- [ ] Use
NoInferto control inference direction
Performance Optimization
- [ ] Reuse Zod schemas (don't create inline)
- [ ] Use
.parse()for known-good data - [ ] Use
.safeParse()for user input - [ ] Enable tRPC batching for multiple queries
- [ ] Cache validation results when appropriate
- [ ] Use Prisma query optimization
- [ ] Avoid N+1 queries with
include - [ ] Use database indexes for common queries
- [ ] Implement pagination for large datasets
- [ ] Use lazy loading for large schemas
Error Handling
- [ ] Provide user-friendly error messages
- [ ] Log detailed errors server-side
- [ ] Send sanitized errors to client
- [ ] Use proper HTTP status codes
- [ ] Implement global error boundary (React)
- [ ] Handle network errors gracefully
- [ ] Show loading/error/success states
- [ ] Retry failed requests with exponential backoff
- [ ] Display validation errors per field
- [ ] Provide actionable error messages
Documentation
Code Documentation
- [ ] Document complex types with JSDoc
- [ ] Add examples to schema definitions
- [ ] Document API endpoints
- [ ] Document error codes and their meanings
- [ ] Create README for shared types
- [ ] Document migration guides
- [ ] Add inline comments for complex logic
- [ ] Generate API documentation from OpenAPI
- [ ] Keep documentation in sync with code
- [ ] Use TypeDoc for type documentation
Developer Experience
- [ ] Set up pre-commit hooks for type checking
- [ ] Run linter in CI/CD
- [ ] Run type checker in CI/CD
- [ ] Provide helpful error messages
- [ ] Create code snippets for common patterns
- [ ] Set up VS Code settings for better DX
- [ ] Use ESLint rules for type safety
- [ ] Configure Prettier for consistent formatting
- [ ] Document common errors and solutions
- [ ] Provide example code in documentation
Maintenance
Regular Maintenance
- [ ] Update dependencies regularly
- [ ] Run
prisma migratefor schema changes - [ ] Regenerate types after backend changes
- [ ] Review and update error messages
- [ ] Audit unused types and schemas
- [ ] Refactor duplicated schemas
- [ ] Monitor bundle size of validation schemas
- [ ] Review and optimize slow validators
- [ ] Update TypeScript to latest stable version
- [ ] Keep Zod/tRPC/Prisma up to date
Breaking Changes
- [ ] Version API endpoints
- [ ] Communicate breaking changes early
- [ ] Provide migration path
- [ ] Deprecate old endpoints before removal
- [ ] Update OpenAPI spec version
- [ ] Test backward compatibility
- [ ] Document breaking changes in CHANGELOG
- [ ] Coordinate backend/frontend deployments
- [ ] Use feature flags for gradual rollout
- [ ] Monitor errors after deployment
Deployment
CI/CD Pipeline
- [ ] Run type checker in CI
- [ ] Run linter in CI
- [ ] Run tests in CI
- [ ] Generate types in CI
- [ ] Build frontend in CI
- [ ] Build backend in CI
- [ ] Run Prisma migrations in CI
- [ ] Deploy with zero downtime
- [ ] Monitor deployment health
- [ ] Rollback on errors
Production Monitoring
- [ ] Log validation errors
- [ ] Monitor API error rates
- [ ] Track response times
- [ ] Monitor database query performance
- [ ] Set up alerts for high error rates
- [ ] Track type errors in production
- [ ] Monitor bundle size
- [ ] Track Core Web Vitals
- [ ] Monitor server resource usage
- [ ] Review logs regularly for issues
---
Progress Tracking:
- [ ] Schema Definition: ___% complete
- [ ] Validation Implementation: ___% complete
- [ ] Type Generation: ___% complete
- [ ] Testing: ___% complete
- [ ] Documentation: ___% complete
- [ ] Overall Type Safety: ___% complete
Target: 100% type coverage across full stack by [DATE]
OrchestKit Type Safety Implementation
How OrchestKit could leverage Zod and end-to-end type safety for the FastAPI backend and React frontend.
Current State
Backend (FastAPI + Pydantic):
- ✅ Runtime validation with Pydantic models
- ✅ OpenAPI schema generation
- ✅ Type hints throughout Python code
Frontend (React + TypeScript):
- ✅ TypeScript for static typing
- ⚠️ Manual type definitions for API responses
- ❌ No runtime validation of API responses
- ❌ Type drift between backend and frontend
Vision: End-to-End Type Safety
Architecture
┌─────────────────────────────────────────────────────┐
│ Backend (FastAPI) │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Pydantic │ ───▶ │ OpenAPI │ │
│ │ Models │ │ Schema │ │
│ └──────────────┘ └──────────────┘ │
│ │ │
└──────────────────────────────┼───────────────────────┘
│
Generate Types
│
▼
┌─────────────────────────────────────────────────────┐
│ Frontend (React + TS) │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Generated TS │ ───▶ │ Zod │ │
│ │ Types │ │ Schemas │ │
│ └──────────────┘ └──────────────┘ │
│ │ │
│ ▼ │
│ Runtime Validation │
└─────────────────────────────────────────────────────┘Implementation Strategies
Strategy 1: OpenAPI → Zod Generation
Use openapi-zod-client to generate Zod schemas from OpenAPI spec:
# Install generator
npm install -D openapi-zod-client
# Generate Zod schemas from OpenAPI spec
npx openapi-zod-client http://localhost:8500/openapi.json -o src/api/generated.tsGenerated output:
// src/api/generated.ts (auto-generated)
import { z } from 'zod'
export const AnalysisSchema = z.object({
id: z.string().uuid(),
url: z.string().url(),
status: z.enum(['pending', 'processing', 'completed', 'failed']),
created_at: z.string().datetime(),
metadata: z.record(z.unknown()).optional(),
})
export type Analysis = z.infer<typeof AnalysisSchema>
export const AnalysisCreateSchema = z.object({
url: z.string().url(),
include_embeddings: z.boolean().default(true),
})
export type AnalysisCreate = z.infer<typeof AnalysisCreateSchema>Usage in React:
// src/features/analysis/hooks/useAnalysis.ts
import { AnalysisSchema } from '@/api/generated'
export function useAnalysis(id: string) {
const { data, error } = useSWR(`/api/v1/analyses/${id}`, async (url) => {
const response = await fetch(url)
const json = await response.json()
// Runtime validation!
const result = AnalysisSchema.safeParse(json)
if (!result.success) {
console.error('Invalid API response:', result.error)
throw new Error('API returned invalid data')
}
return result.data
})
return { data, error }
}Strategy 2: Pydantic ↔ Zod Pattern Mapping
Manual mapping between Pydantic and Zod patterns:
Backend (Pydantic):
# backend/app/schemas/analysis.py
from pydantic import BaseModel, Field, HttpUrl
from datetime import datetime
from enum import Enum
class AnalysisStatus(str, Enum):
PENDING = "pending"
PROCESSING = "processing"
COMPLETED = "completed"
FAILED = "failed"
class AnalysisCreate(BaseModel):
url: HttpUrl
include_embeddings: bool = True
class Analysis(BaseModel):
id: str = Field(..., description="UUID")
url: HttpUrl
status: AnalysisStatus
created_at: datetime
metadata: dict[str, Any] | None = None
class Config:
from_attributes = TrueFrontend (Zod):
// frontend/src/schemas/analysis.ts
import { z } from 'zod'
export const AnalysisStatusSchema = z.enum([
'pending',
'processing',
'completed',
'failed'
])
export const AnalysisCreateSchema = z.object({
url: z.string().url(),
include_embeddings: z.boolean().default(true),
})
export const AnalysisSchema = z.object({
id: z.string().uuid(),
url: z.string().url(),
status: AnalysisStatusSchema,
created_at: z.string().datetime().transform(s => new Date(s)),
metadata: z.record(z.unknown()).nullable(),
})
export type AnalysisStatus = z.infer<typeof AnalysisStatusSchema>
export type AnalysisCreate = z.infer<typeof AnalysisCreateSchema>
export type Analysis = z.infer<typeof AnalysisSchema>Strategy 3: Shared Type Definitions
Generate TypeScript types from Pydantic, then create Zod schemas:
# Generate TypeScript types from OpenAPI
npx openapi-typescript http://localhost:8500/openapi.json -o src/api/types.tsThen create Zod schemas that satisfy the generated types:
// src/api/schemas.ts
import { z } from 'zod'
import type { components } from './types' // Generated types
export const AnalysisSchema = z.object({
id: z.string().uuid(),
url: z.string().url(),
status: z.enum(['pending', 'processing', 'completed', 'failed']),
created_at: z.string().datetime(),
metadata: z.record(z.unknown()).nullable(),
}) satisfies z.ZodType<components['schemas']['Analysis']>
// TypeScript ensures Zod schema matches OpenAPI type!Real-World Example: Analysis Submission
Backend (FastAPI)
# backend/app/api/v1/analyses.py
from fastapi import APIRouter, Depends
from app.schemas.analysis import AnalysisCreate, Analysis
from app.services.analysis_service import AnalysisService
router = APIRouter()
@router.post("/", response_model=Analysis, status_code=201)
async def create_analysis(
data: AnalysisCreate,
service: AnalysisService = Depends()
) -> Analysis:
"""Create new analysis with validation."""
return await service.create(data)Frontend (React + Zod)
// src/features/analysis/components/AnalysisForm.tsx
'use client'
import { z } from 'zod'
import { useForm } from 'react-hook-form'
import { zodResolver } from '@hookform/resolvers/zod'
import { AnalysisCreateSchema, AnalysisSchema } from '@/schemas/analysis'
export function AnalysisForm() {
const form = useForm<z.infer<typeof AnalysisCreateSchema>>({
resolver: zodResolver(AnalysisCreateSchema),
defaultValues: {
include_embeddings: true,
},
})
const onSubmit = async (data: z.infer<typeof AnalysisCreateSchema>) => {
try {
const response = await fetch('/api/v1/analyses/', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(data),
})
const json = await response.json()
// Runtime validation of API response!
const result = AnalysisSchema.safeParse(json)
if (!result.success) {
console.error('Invalid API response:', result.error)
throw new Error('Server returned invalid data')
}
// result.data is fully typed!
console.log('Created analysis:', result.data.id)
} catch (error) {
form.setError('root', {
message: error instanceof Error ? error.message : 'Failed to create analysis'
})
}
}
return (
<form onSubmit={form.handleSubmit(onSubmit)}>
<input {...form.register('url')} placeholder="Enter URL" />
{form.formState.errors.url && (
<span className="error">{form.formState.errors.url.message}</span>
)}
<label>
<input type="checkbox" {...form.register('include_embeddings')} />
Include embeddings
</label>
<button type="submit" disabled={form.formState.isSubmitting}>
{form.formState.isSubmitting ? 'Creating...' : 'Create Analysis'}
</button>
</form>
)
}SSE (Server-Sent Events) Validation
OrchestKit uses SSE for real-time progress updates. Validate event payloads:
Backend:
# backend/app/schemas/events.py
from pydantic import BaseModel
class ProgressEvent(BaseModel):
type: Literal["progress"]
agent_id: str
stage: str
progress: float # 0.0 to 1.0
message: str
class ErrorEvent(BaseModel):
type: Literal["error"]
error: str
details: dict[str, Any] | None = NoneFrontend:
// src/schemas/events.ts
import { z } from 'zod'
export const ProgressEventSchema = z.object({
type: z.literal('progress'),
agent_id: z.string(),
stage: z.string(),
progress: z.number().min(0).max(1),
message: z.string(),
})
export const ErrorEventSchema = z.object({
type: z.literal('error'),
error: z.string(),
details: z.record(z.unknown()).optional(),
})
export const EventSchema = z.discriminatedUnion('type', [
ProgressEventSchema,
ErrorEventSchema,
])
export type Event = z.infer<typeof EventSchema>
// src/hooks/useAnalysisProgress.ts
export function useAnalysisProgress(id: string) {
const [events, setEvents] = useState<Event[]>([])
useEffect(() => {
const eventSource = new EventSource(`/api/v1/analyses/${id}/progress`)
eventSource.onmessage = (event) => {
const json = JSON.parse(event.data)
// Runtime validation!
const result = EventSchema.safeParse(json)
if (result.success) {
setEvents(prev => [...prev, result.data])
} else {
console.error('Invalid SSE event:', result.error)
}
}
return () => eventSource.close()
}, [id])
return events
}Benefits for OrchestKit
1. Type Safety Across Stack
- Backend: Pydantic validates inputs
- Frontend: Zod validates API responses
- No runtime surprises from mismatched data
2. Single Source of Truth
- OpenAPI spec generated from Pydantic
- TypeScript types + Zod schemas generated from OpenAPI
- Changes to backend models automatically update frontend
3. Better DX
- Autocomplete for API responses
- Compile-time errors when API changes
- Runtime validation catches issues early
4. Form Validation
- Use same Zod schemas for form validation (react-hook-form)
- Consistent validation rules
- Better error messages
5. Testing
- Mock data generators from Zod schemas
- Type-safe test fixtures
- Validate test data matches production
Migration Path
Phase 1: Add Zod to Frontend
npm install zod @hookform/resolversPhase 2: Create Schemas for Critical Types
Start with most-used types (Analysis, Artifact, Chunk):
// src/schemas/index.ts
export * from './analysis'
export * from './artifact'
export * from './chunk'Phase 3: Add Runtime Validation to API Calls
Wrap fetch calls with validation:
// src/lib/api-client.ts
export async function apiRequest<T>(
url: string,
schema: z.ZodType<T>,
options?: RequestInit
): Promise<T> {
const response = await fetch(url, options)
const json = await response.json()
const result = schema.safeParse(json)
if (!result.success) {
console.error('API validation failed:', result.error)
throw new Error('Invalid API response')
}
return result.data
}Phase 4: Automate Type Generation
Set up CI/CD to regenerate types on backend changes:
# .github/workflows/generate-types.yml
name: Generate Frontend Types
on:
push:
paths:
- 'backend/app/schemas/**'
jobs:
generate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- run: npm run generate:types
- uses: peter-evans/create-pull-request@v5
with:
title: "Update generated API types"Pydantic ↔ Zod Cheat Sheet
| Pydantic (Python) | Zod (TypeScript) |
|---|---|
str | z.string() |
int | z.number().int() |
float | z.number() |
bool | z.boolean() |
datetime | z.date() or z.string().datetime() |
HttpUrl | z.string().url() |
EmailStr | z.string().email() |
Field(..., min_length=1) | z.string().min(1) |
Field(..., max_length=100) | z.string().max(100) |
Field(..., ge=0, le=100) | z.number().min(0).max(100) |
list[str] | z.array(z.string()) |
dict[str, Any] | z.record(z.unknown()) |
| `str \ | None` |
Literal["active"] | z.literal("active") |
Enum | z.enum([...]) or z.nativeEnum(...) |
@validator | .refine(...) or .transform(...) |
BaseModel | z.object({...}) |
Conclusion
Adding Zod to OrchestKit's frontend would provide:
- ✅ Runtime safety for API responses
- ✅ Form validation with same schemas
- ✅ Type inference for better DX
- ✅ Single source of truth via OpenAPI
- ✅ Gradual migration path
Start with critical types, then expand coverage over time.
Prisma Types and Type-Safe Queries
Complete guide to leveraging Prisma's generated TypeScript types for type-safe database access.
Prisma Schema Basics
// schema.prisma
generator client {
provider = "prisma-client-js"
previewFeatures = ["fullTextSearch", "postgresqlExtensions"]
}
datasource db {
provider = "postgresql"
url = env("DATABASE_URL")
}
model User {
id String @id @default(cuid())
email String @unique
name String?
posts Post[]
profile Profile?
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
@@index([email])
}
model Post {
id String @id @default(cuid())
title String
content String?
published Boolean @default(false)
author User @relation(fields: [authorId], references: [id], onDelete: Cascade)
authorId String
tags Tag[]
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
@@index([authorId])
@@index([published])
}
model Profile {
id String @id @default(cuid())
bio String?
avatar String?
user User @relation(fields: [userId], references: [id])
userId String @unique
}
model Tag {
id String @id @default(cuid())
name String @unique
posts Post[]
}Generated Types
After running npx prisma generate, Prisma creates TypeScript types:
import { PrismaClient, User, Post, Prisma } from '@prisma/client'
// Model types (from database schema)
const user: User = {
id: 'abc123',
email: 'user@example.com',
name: 'John Doe',
createdAt: new Date(),
updatedAt: new Date()
}
// Type-safe client
const prisma = new PrismaClient()
// All queries are fully typed!
const posts = await prisma.post.findMany()
// ^? Post[]
const post = await prisma.post.findUnique({ where: { id: '123' } })
// ^? Post | nullType-Safe Queries
Basic CRUD Operations
// Create
const newUser = await prisma.user.create({
data: {
email: 'new@example.com',
name: 'New User',
posts: {
create: [
{ title: 'First Post', content: 'Hello world' }
]
}
}
})
// newUser: User
// Read
const user = await prisma.user.findUnique({
where: { email: 'user@example.com' }
})
// user: User | null
const users = await prisma.user.findMany({
where: {
posts: {
some: {
published: true
}
}
}
})
// users: User[]
// Update
const updated = await prisma.user.update({
where: { id: 'abc123' },
data: { name: 'Updated Name' }
})
// updated: User
// Delete
const deleted = await prisma.user.delete({
where: { id: 'abc123' }
})
// deleted: UserIncludes and Selects
// Include relations
const userWithPosts = await prisma.user.findUnique({
where: { id: 'abc123' },
include: {
posts: true,
profile: true
}
})
// userWithPosts: User & { posts: Post[], profile: Profile | null } | null
// Nested includes
const userWithPublishedPosts = await prisma.user.findUnique({
where: { id: 'abc123' },
include: {
posts: {
where: { published: true },
include: {
tags: true
}
}
}
})
// userWithPublishedPosts: User & { posts: (Post & { tags: Tag[] })[] } | null
// Select specific fields
const userEmail = await prisma.user.findUnique({
where: { id: 'abc123' },
select: {
email: true,
name: true
}
})
// userEmail: { email: string, name: string | null } | null
// Mix select and include (not allowed, pick one!)
// ❌ This is a compile-time error:
// const bad = await prisma.user.findUnique({
// where: { id: 'abc123' },
// select: { email: true },
// include: { posts: true } // ERROR: Cannot use both!
// })Prisma Types Namespace
import { Prisma } from '@prisma/client'
// Input types for create/update
type UserCreateInput = Prisma.UserCreateInput
const newUser: UserCreateInput = {
email: 'user@example.com',
name: 'John',
posts: {
create: [{ title: 'Post', content: 'Content' }]
}
}
// Where input for filtering
type UserWhereInput = Prisma.UserWhereInput
const filter: UserWhereInput = {
email: { contains: '@example.com' },
posts: {
some: { published: true }
}
}
// Order by input
type UserOrderByInput = Prisma.UserOrderByWithRelationInput
const orderBy: UserOrderByInput = {
createdAt: 'desc',
posts: {
_count: 'desc' // Order by number of posts
}
}
// Select input
type UserSelectInput = Prisma.UserSelect
const select: UserSelectInput = {
id: true,
email: true,
posts: {
select: {
title: true
}
}
}Custom Type Helpers
Get Return Types
import { Prisma } from '@prisma/client'
// Type of findUnique result with includes
type UserWithPosts = Prisma.UserGetPayload<{
include: { posts: true }
}>
// = User & { posts: Post[] }
// Type of findMany result with select
type UserEmailOnly = Prisma.UserGetPayload<{
select: { email: true, name: true }
}>
// = { email: string, name: string | null }
// Use in functions
async function getUserWithPosts(id: string): Promise<UserWithPosts | null> {
return await prisma.user.findUnique({
where: { id },
include: { posts: true }
})
}Validator Pattern
import { Prisma } from '@prisma/client'
// Define reusable query options
const userWithPostsArgs = Prisma.validator<Prisma.UserDefaultArgs>()({
include: { posts: true }
})
// Get type from validator
type UserWithPosts = Prisma.UserGetPayload<typeof userWithPostsArgs>
// Use in multiple places
async function getUser(id: string): Promise<UserWithPosts | null> {
return await prisma.user.findUnique({
where: { id },
...userWithPostsArgs
})
}Extending Prisma Types
Add Custom Fields
import { User, Post } from '@prisma/client'
// Extend with computed fields
interface UserWithPostCount extends User {
postCount: number
}
async function getUserWithPostCount(id: string): Promise<UserWithPostCount | null> {
const user = await prisma.user.findUnique({
where: { id },
include: {
_count: {
select: { posts: true }
}
}
})
if (!user) return null
return {
...user,
postCount: user._count.posts
}
}
// Extend with virtual fields
interface PostWithSlug extends Post {
slug: string
}
function addSlug(post: Post): PostWithSlug {
return {
...post,
slug: post.title.toLowerCase().replace(/\s+/g, '-')
}
}Partial Models
import { Prisma } from '@prisma/client'
// Pick specific fields
type UserPublicInfo = Pick<User, 'id' | 'name' | 'email'>
// Omit sensitive fields
type UserSafe = Omit<User, 'password' | 'resetToken'>
// Make fields optional
type UserUpdate = Partial<Pick<User, 'name' | 'email'>>
// Combine with Prisma types
type UserCreateDTO = Omit<Prisma.UserCreateInput, 'id' | 'createdAt' | 'updatedAt'>Type-Safe Transactions
// Simple transaction
const [user, post] = await prisma.$transaction([
prisma.user.create({ data: { email: 'user@example.com' } }),
prisma.post.create({ data: { title: 'Post', authorId: 'userId' } })
])
// user: User, post: Post
// Interactive transaction
const result = await prisma.$transaction(async (tx) => {
const user = await tx.user.create({
data: { email: 'user@example.com' }
})
const post = await tx.post.create({
data: {
title: 'First Post',
authorId: user.id
}
})
return { user, post }
})
// result: { user: User, post: Post }
// With isolation level
await prisma.$transaction(async (tx) => {
// Your queries here
}, {
isolationLevel: Prisma.TransactionIsolationLevel.Serializable,
maxWait: 5000,
timeout: 10000
})Raw Queries with Types
import { Prisma } from '@prisma/client'
// Type-safe raw query
const users = await prisma.$queryRaw<User[]>`
SELECT * FROM "User" WHERE email LIKE ${`%@example.com`}
`
// users: User[]
// With custom type
interface UserWithCount {
id: string
email: string
postCount: bigint
}
const usersWithCounts = await prisma.$queryRaw<UserWithCount[]>`
SELECT u.id, u.email, COUNT(p.id) as "postCount"
FROM "User" u
LEFT JOIN "Post" p ON p."authorId" = u.id
GROUP BY u.id
`
// Execute raw (no return value)
await prisma.$executeRaw`
UPDATE "User" SET name = 'Updated' WHERE id = ${userId}
`
// Use Prisma.sql for safer raw queries
const email = 'user@example.com'
const users = await prisma.$queryRaw<User[]>(
Prisma.sql`SELECT * FROM "User" WHERE email = ${email}`
)Prisma with Zod
Combine Prisma types with Zod validation:
import { z } from 'zod'
import { Prisma } from '@prisma/client'
// Create Zod schema matching Prisma model
const UserCreateSchema = z.object({
email: z.string().email(),
name: z.string().min(1).max(100).optional(),
posts: z.array(z.object({
title: z.string().min(1),
content: z.string().optional()
})).optional()
}) satisfies z.ZodType<Prisma.UserCreateInput>
// Use in API
export async function createUser(input: unknown) {
const validated = UserCreateSchema.parse(input)
return await prisma.user.create({
data: validated
})
}
// Generate Zod from Prisma schema automatically
// npm install zod-prisma-types
// Then add to schema.prisma:
// generator zod {
// provider = "zod-prisma-types"
// }Repository Pattern
// repositories/user.repository.ts
import { PrismaClient, User, Prisma } from '@prisma/client'
export class UserRepository {
constructor(private prisma: PrismaClient) {}
async findById(id: string): Promise<User | null> {
return await this.prisma.user.findUnique({ where: { id } })
}
async findByEmail(email: string): Promise<User | null> {
return await this.prisma.user.findUnique({ where: { email } })
}
async create(data: Prisma.UserCreateInput): Promise<User> {
return await this.prisma.user.create({ data })
}
async update(id: string, data: Prisma.UserUpdateInput): Promise<User> {
return await this.prisma.user.update({ where: { id }, data })
}
async delete(id: string): Promise<User> {
return await this.prisma.user.delete({ where: { id } })
}
// Custom queries with proper types
async findUsersWithPublishedPosts(): Promise<Array<User & { posts: Post[] }>> {
return await this.prisma.user.findMany({
where: {
posts: {
some: { published: true }
}
},
include: {
posts: {
where: { published: true }
}
}
})
}
}
// Usage
const userRepo = new UserRepository(prisma)
const user = await userRepo.findById('123')
// user: User | nullBest Practices
1. Always run `prisma generate` after schema changes 2. Use strict mode in tsconfig.json 3. Leverage `Prisma.validator` for reusable query options 4. Use `Prisma.UserGetPayload` to extract types from queries 5. Combine with Zod for input validation 6. Use repositories to encapsulate database logic 7. Enable preview features carefully in production 8. Type transactions properly for complex operations 9. Avoid `any` types - Prisma provides full type coverage 10. Use raw queries sparingly - lose some type safety
tRPC Setup and Patterns
Complete guide to building type-safe APIs with tRPC v11+.
Core Concepts
tRPC provides end-to-end type safety between server and client without code generation:
- Routers: Group related procedures
- Procedures: Individual API endpoints (query/mutation/subscription)
- Context: Shared data across procedures (auth, db, etc.)
- Middleware: Logic that runs before procedures
Server Setup
Initialize tRPC
// server/trpc.ts
import { initTRPC, TRPCError } from '@trpc/server'
import type { CreateNextContextOptions } from '@trpc/server/adapters/next'
import { ZodError } from 'zod'
import superjson from 'superjson'
// Context type
export interface Context {
userId?: string
db: PrismaClient
req: Request
}
// Create context from request
export async function createContext(
opts: CreateNextContextOptions
): Promise<Context> {
const token = opts.req.headers.get('authorization')?.replace('Bearer ', '')
const userId = token ? await verifyToken(token) : undefined
return {
userId,
db: prisma,
req: opts.req
}
}
// Initialize tRPC with context type
const t = initTRPC.context<Context>().create({
transformer: superjson, // Serialize Date, Map, Set, etc.
errorFormatter({ shape, error }) {
return {
...shape,
data: {
...shape.data,
zodError:
error.cause instanceof ZodError
? error.cause.flatten()
: null,
},
}
},
})
// Export reusable pieces
export const router = t.router
export const publicProcedure = t.procedure
export const middleware = t.middlewareMiddleware for Authentication
// server/trpc.ts (continued)
// Auth middleware
const enforceAuth = middleware(async ({ ctx, next }) => {
if (!ctx.userId) {
throw new TRPCError({
code: 'UNAUTHORIZED',
message: 'You must be logged in'
})
}
return next({
ctx: {
userId: ctx.userId, // Now guaranteed non-null
},
})
})
// Protected procedure (requires auth)
export const protectedProcedure = publicProcedure.use(enforceAuth)
// Rate limiting middleware
const rateLimit = middleware(async ({ ctx, next, path }) => {
const key = `rate-limit:${ctx.userId || 'anon'}:${path}`
const count = await redis.incr(key)
if (count === 1) {
await redis.expire(key, 60) // 60 second window
}
if (count > 100) {
throw new TRPCError({
code: 'TOO_MANY_REQUESTS',
message: 'Rate limit exceeded'
})
}
return next()
})
export const rateLimitedProcedure = publicProcedure.use(rateLimit)Router Structure
Basic Router
// server/routers/_app.ts
import { router, publicProcedure, protectedProcedure } from '../trpc'
import { z } from 'zod'
import { userRouter } from './user'
import { postRouter } from './post'
export const appRouter = router({
// Inline procedures
hello: publicProcedure
.input(z.object({ name: z.string() }))
.query(({ input }) => {
return { greeting: `Hello ${input.name}!` }
}),
// Nested routers
user: userRouter,
post: postRouter,
})
// Export type for client
export type AppRouter = typeof appRouterNested Router Example
// server/routers/post.ts
import { router, publicProcedure, protectedProcedure } from '../trpc'
import { z } from 'zod'
import { TRPCError } from '@trpc/server'
const PostInputSchema = z.object({
title: z.string().min(1).max(200),
content: z.string().optional(),
published: z.boolean().default(false),
tags: z.array(z.string()).max(10)
})
export const postRouter = router({
// List posts with cursor pagination
list: publicProcedure
.input(z.object({
limit: z.number().min(1).max(100).default(10),
cursor: z.string().optional(),
published: z.boolean().optional()
}))
.query(async ({ ctx, input }) => {
const posts = await ctx.db.post.findMany({
take: input.limit + 1,
cursor: input.cursor ? { id: input.cursor } : undefined,
where: { published: input.published },
orderBy: { createdAt: 'desc' },
include: {
author: {
select: { id: true, name: true }
}
}
})
let nextCursor: string | undefined
if (posts.length > input.limit) {
const nextItem = posts.pop()
nextCursor = nextItem!.id
}
return {
items: posts,
nextCursor
}
}),
// Get single post
byId: publicProcedure
.input(z.object({ id: z.string() }))
.query(async ({ ctx, input }) => {
const post = await ctx.db.post.findUnique({
where: { id: input.id },
include: {
author: true,
comments: {
orderBy: { createdAt: 'desc' }
}
}
})
if (!post) {
throw new TRPCError({
code: 'NOT_FOUND',
message: 'Post not found'
})
}
return post
}),
// Create post (protected)
create: protectedProcedure
.input(PostInputSchema)
.mutation(async ({ ctx, input }) => {
return await ctx.db.post.create({
data: {
...input,
authorId: ctx.userId
}
})
}),
// Update post (protected)
update: protectedProcedure
.input(z.object({
id: z.string(),
data: PostInputSchema.partial()
}))
.mutation(async ({ ctx, input }) => {
// Check ownership
const post = await ctx.db.post.findUnique({
where: { id: input.id }
})
if (!post || post.authorId !== ctx.userId) {
throw new TRPCError({
code: 'FORBIDDEN',
message: 'Cannot update this post'
})
}
return await ctx.db.post.update({
where: { id: input.id },
data: input.data
})
}),
// Delete post (protected)
delete: protectedProcedure
.input(z.object({ id: z.string() }))
.mutation(async ({ ctx, input }) => {
// Check ownership
const post = await ctx.db.post.findUnique({
where: { id: input.id }
})
if (!post || post.authorId !== ctx.userId) {
throw new TRPCError({
code: 'FORBIDDEN',
message: 'Cannot delete this post'
})
}
await ctx.db.post.delete({
where: { id: input.id }
})
return { success: true }
})
})Client Setup
Next.js App Router Setup
// app/_trpc/client.ts
import { createTRPCReact } from '@trpc/react-query'
import type { AppRouter } from '@/server/routers/_app'
export const trpc = createTRPCReact<AppRouter>()
// app/_trpc/Provider.tsx
'use client'
import { QueryClient, QueryClientProvider } from '@tanstack/react-query'
import { httpBatchLink } from '@trpc/client'
import { useState } from 'react'
import superjson from 'superjson'
import { trpc } from './client'
export function TRPCProvider({ children }: { children: React.ReactNode }) {
const [queryClient] = useState(() => new QueryClient({
defaultOptions: {
queries: {
staleTime: 5 * 1000, // 5 seconds
refetchOnWindowFocus: false
}
}
}))
const [trpcClient] = useState(() =>
trpc.createClient({
transformer: superjson,
links: [
httpBatchLink({
url: '/api/trpc',
headers() {
const token = localStorage.getItem('token')
return token ? { authorization: `Bearer ${token}` } : {}
}
})
]
})
)
return (
<trpc.Provider client={trpcClient} queryClient={queryClient}>
<QueryClientProvider client={queryClient}>
{children}
</QueryClientProvider>
</trpc.Provider>
)
}Next.js API Handler
// app/api/trpc/[trpc]/route.ts
import { fetchRequestHandler } from '@trpc/server/adapters/fetch'
import { appRouter } from '@/server/routers/_app'
import { createContext } from '@/server/trpc'
const handler = (req: Request) =>
fetchRequestHandler({
endpoint: '/api/trpc',
req,
router: appRouter,
createContext,
})
export { handler as GET, handler as POST }Client Usage Patterns
Queries
'use client'
import { trpc } from '@/app/_trpc/client'
export function PostList() {
// Basic query
const { data, isLoading, error } = trpc.post.list.useQuery({
limit: 10,
published: true
})
// Query with options
const { data: post } = trpc.post.byId.useQuery(
{ id: postId },
{
enabled: !!postId, // Only run when postId exists
refetchInterval: 5000, // Refetch every 5s
retry: 3
}
)
// Infinite query (cursor pagination)
const {
data,
fetchNextPage,
hasNextPage,
isFetchingNextPage
} = trpc.post.list.useInfiniteQuery(
{ limit: 10 },
{
getNextPageParam: (lastPage) => lastPage.nextCursor
}
)
if (isLoading) return <div>Loading...</div>
if (error) return <div>Error: {error.message}</div>
return (
<div>
{data?.items.map(post => (
<div key={post.id}>{post.title}</div>
))}
</div>
)
}Mutations
'use client'
import { trpc } from '@/app/_trpc/client'
export function CreatePostForm() {
const utils = trpc.useUtils()
const createPost = trpc.post.create.useMutation({
// Optimistic update
onMutate: async (newPost) => {
await utils.post.list.cancel()
const previous = utils.post.list.getData()
utils.post.list.setData(
{ limit: 10 },
(old) => old ? {
...old,
items: [newPost as any, ...old.items]
} : old
)
return { previous }
},
// Revert on error
onError: (err, newPost, context) => {
utils.post.list.setData({ limit: 10 }, context?.previous)
},
// Refetch on success
onSuccess: () => {
utils.post.list.invalidate()
}
})
const handleSubmit = (data: FormData) => {
createPost.mutate({
title: data.get('title') as string,
content: data.get('content') as string,
published: false,
tags: []
})
}
return (
<form action={handleSubmit}>
<input name="title" required />
<textarea name="content" />
<button type="submit" disabled={createPost.isPending}>
{createPost.isPending ? 'Creating...' : 'Create Post'}
</button>
{createPost.error && (
<div>Error: {createPost.error.message}</div>
)}
</form>
)
}Server Components (Next.js)
// app/posts/page.tsx
import { createCaller } from '@/server/routers/_app'
import { createContext } from '@/server/trpc'
export default async function PostsPage() {
// Create server-side caller
const caller = createCaller(await createContext({
req: new Request('http://localhost:3000')
} as any))
// Directly call procedures
const posts = await caller.post.list({ limit: 10 })
return (
<div>
{posts.items.map(post => (
<div key={post.id}>{post.title}</div>
))}
</div>
)
}Advanced Patterns
Subscriptions (WebSocket)
// Server
export const postRouter = router({
onNewPost: publicProcedure
.subscription(() => {
return observable<Post>((emit) => {
const listener = (post: Post) => emit.next(post)
eventEmitter.on('post:created', listener)
return () => {
eventEmitter.off('post:created', listener)
}
})
})
})
// Client
const { data } = trpc.post.onNewPost.useSubscription(undefined, {
onData(post) {
console.log('New post:', post)
}
})Batching
// Automatic request batching
const [user, posts, comments] = await Promise.all([
trpc.user.byId.query({ id: '1' }),
trpc.post.list.query({ limit: 10 }),
trpc.comment.recent.query({ limit: 5 })
])
// Sent as single HTTP request!Prefetching
// Prefetch in server component
export default async function Page() {
const caller = createCaller(await createContext(...))
await Promise.all([
caller.post.list.prefetch({ limit: 10 }),
caller.user.me.prefetch()
])
return <HydrateClient><ClientComponent /></HydrateClient>
}Error Handling
// Custom error types
class CustomTRPCError extends TRPCError {
constructor(message: string, code: TRPC_ERROR_CODE_KEY = 'INTERNAL_SERVER_ERROR') {
super({ code, message })
}
}
// Client-side error handling
const { error } = trpc.post.create.useMutation()
if (error) {
if (error.data?.code === 'UNAUTHORIZED') {
router.push('/login')
} else if (error.data?.zodError) {
// Handle validation errors
const fieldErrors = error.data.zodError.fieldErrors
} else {
toast.error(error.message)
}
}Best Practices
1. Use nested routers - Organize by domain (user, post, comment) 2. Validate all inputs - Use Zod schemas for type safety + runtime validation 3. Use middleware - DRY auth, logging, rate limiting 4. Enable batching - Combine multiple requests into one 5. Use superjson - Serialize Date, Map, Set automatically 6. Type-safe errors - Format Zod errors in errorFormatter 7. Optimistic updates - Better UX with immediate feedback 8. Prefetch data - Faster navigation with server prefetching
Ty Type Checker Patterns (Rust-Based Python Type Checking)
Overview
Ty is a Rust-based static type checker for Python that enforces stricter type safety than mypy. It requires explicit type annotations and runtime checks for type narrowing.
Key Difference from mypy: Ty cannot narrow types through simple conditionals - it requires explicit isinstance() checks and type annotations.
Pattern: Safe Optional Extraction from Dicts
Problem
When extracting values from dictionaries (e.g., agent results, API responses) for database storage, ty's type checker needs explicit help to understand type narrowing.
# ❌ FAILS with ty - type checker can't narrow
result = {"findings": {...}, "confidence_score": 0.85}
findings_raw = result.get("findings", {})
confidence_raw = result.get("confidence_score")
# ty sees: object | None (can't narrow)
findings_to_save = findings_raw if isinstance(findings_raw, dict) else None
confidence_to_save = float(confidence_raw) if confidence_raw is not None else None
# Error: confidence_raw could be non-numeric!Solution: Explicit Type Annotations + isinstance Checks
from typing import cast
# Extract from result dict
findings_raw = result.get("findings", {})
confidence_raw = result.get("confidence_score")
# Type-safe extraction with explicit annotations
findings_to_save: dict[str, object] | None = (
cast("dict[str, object]", findings_raw) if isinstance(findings_raw, dict) else None
)
confidence_to_save: float | None = (
float(confidence_raw) if isinstance(confidence_raw, (int, float)) else None
)Why This Works
1. Explicit type annotation (findings_to_save: dict[str, object] | None) tells ty the expected type 2. isinstance() runtime check proves to ty that the value is the expected type 3. cast() bridges the gap between runtime check and compile-time type 4. Numeric type check (isinstance(x, (int, float))) ensures float() won't fail
Real-World Example: Agent Result Processing
from typing import Any, cast
async def save_agent_result_to_db(
agent_result: dict[str, Any],
db_repo: AgentResultRepository
) -> None:
"""
Save agent result to database with ty-compliant type safety.
Args:
agent_result: Raw dict from agent execution (untyped)
db_repo: Database repository for persistence
"""
# Extract raw values (ty sees these as object | None)
findings_raw = agent_result.get("findings", {})
confidence_raw = agent_result.get("confidence_score")
metadata_raw = agent_result.get("metadata", {})
tags_raw = agent_result.get("tags", [])
# Type-safe extraction with explicit annotations
findings: dict[str, object] | None = (
cast("dict[str, object]", findings_raw)
if isinstance(findings_raw, dict)
else None
)
confidence: float | None = (
float(confidence_raw)
if isinstance(confidence_raw, (int, float))
else None
)
metadata: dict[str, object] | None = (
cast("dict[str, object]", metadata_raw)
if isinstance(metadata_raw, dict)
else None
)
tags: list[str] | None = (
cast("list[str]", tags_raw)
if isinstance(tags_raw, list) and all(isinstance(t, str) for t in tags_raw)
else None
)
# Now db_repo methods receive properly typed values
await db_repo.create(
findings=findings,
confidence_score=confidence,
metadata=metadata,
tags=tags
)Pattern: Handling Mixed Numeric Types
Problem
LLM responses or API results may return numeric values as strings, ints, or floats.
# ❌ FAILS - ty can't guarantee str is numeric
score_raw = result.get("score") # Could be "8.5", 8.5, or 8
score: float = float(score_raw) # Error: score_raw could be None or non-numericSolution: Defensive Type Checking
score_raw = result.get("score")
# Option 1: Safe conversion with fallback
score: float | None = None
if isinstance(score_raw, (int, float)):
score = float(score_raw)
elif isinstance(score_raw, str):
try:
score = float(score_raw)
except ValueError:
score = None
# Option 2: Inline with explicit annotation
score: float | None = (
float(score_raw)
if isinstance(score_raw, (int, float, str)) and (
isinstance(score_raw, (int, float)) or score_raw.replace(".", "", 1).isdigit()
)
else None
)Pattern: List Type Narrowing
Problem
Lists from untyped sources need element-level validation.
# ❌ FAILS - ty can't guarantee list elements are strings
tags_raw = result.get("tags", [])
tags: list[str] = tags_raw # Error: could be list[Any]Solution: Element-Level isinstance Check
from typing import cast
tags_raw = result.get("tags", [])
# Validate all elements are strings
tags: list[str] | None = (
cast("list[str]", tags_raw)
if isinstance(tags_raw, list) and all(isinstance(t, str) for t in tags_raw)
else None
)
# Alternative: Filter out non-strings
tags_filtered: list[str] = [
t for t in tags_raw if isinstance(t, str)
] if isinstance(tags_raw, list) else []Pattern: Nested Dict Extraction
Problem
Nested dictionaries require multiple levels of type checking.
# ❌ FAILS - ty can't narrow nested access
config_raw = data.get("config", {})
timeout = config_raw.get("timeout", 30) # Error: config_raw could be NoneSolution: Cascading isinstance Checks
from typing import cast
config_raw = data.get("config", {})
# Safe nested access
timeout: int = 30 # Default
if isinstance(config_raw, dict):
timeout_raw = config_raw.get("timeout")
if isinstance(timeout_raw, int):
timeout = timeout_raw
elif isinstance(timeout_raw, str) and timeout_raw.isdigit():
timeout = int(timeout_raw)
# Alternative: One-liner with explicit annotation
timeout: int = (
int(cast("dict[str, Any]", config_raw).get("timeout", 30))
if isinstance(config_raw, dict) and isinstance(config_raw.get("timeout"), int)
else 30
)When to Use These Patterns
Use explicit type annotations + isinstance checks when:
- Extracting values from
dict[str, Any]ordict[str, object] - Processing LLM/API responses with unknown structure
- Converting between JSON and database models
- Handling optional numeric types from external sources
- Working with untyped third-party libraries
Key Principle: Ty requires proof of type safety. Provide that proof through: 1. Explicit type annotations (: type | None) 2. Runtime type checks (isinstance(x, type)) 3. cast() for bridging runtime → compile-time 4. Defensive conversions with try/except for strings
Comparison: mypy vs ty
# mypy (lenient) - both pass
x = data.get("value")
y = float(x) if x is not None else None
# ty (strict) - first fails, second passes
x = data.get("value")
y = float(x) if x is not None else None # ❌ x could be non-numeric
x = data.get("value")
y: float | None = (
float(x) if isinstance(x, (int, float)) else None # ✅ Explicit check
)Resources
- Ty documentation: Coming soon (Rust-based type checker for Python)
- Related skill:
references/typescript-5-features.md(similar strict patterns) - OrchestKit usage: Backend agent result processing (
backend/app/workflows/nodes/)
TypeScript 5.x Features for Type Safety
Modern TypeScript features (5.0-5.7) for maximum type safety in 2026 applications.
TypeScript 5.7 Features (Latest)
Path Rewriting for Relative Imports
// tsconfig.json
{
"compilerOptions": {
"module": "nodenext",
"rewriteRelativeImportExtensions": true
}
}
// Write this:
import { User } from './user.ts'
// Compiles to:
import { User } from './user.js'Checked Imports
// Prevent imports from test files in production
// tsconfig.json
{
"compilerOptions": {
"allowImportsFromTest": false // New in 5.7
}
}Nullish and Truthy Checks
// Better error messages for nullish checks
function process(value: string | null | undefined) {
if (value) { // TS 5.7 warns about implicit coercion
return value.toUpperCase()
}
}
// Better:
function process(value: string | null | undefined) {
if (value != null) { // Explicit nullish check
return value.toUpperCase()
}
}TypeScript 5.5 Features
Inferred Type Predicates
// TS 5.5+ can infer type predicates automatically!
function isString(value: unknown) {
return typeof value === 'string'
}
// isString is inferred as: (value: unknown) => value is string
const values: unknown[] = ['hello', 42, 'world']
const strings = values.filter(isString)
// ^? string[] (automatically narrowed!)
// Before TS 5.5, you had to write:
function isString(value: unknown): value is string {
return typeof value === 'string'
}Const Type Parameters
// Preserve literal types in generic functions
function identity<const T>(value: T): T {
return value
}
const result = identity({ x: 10, y: 20 })
// ^? { readonly x: 10, readonly y: 20 }
// Not: { x: number, y: number }
// Useful for config objects
function createConfig<const T extends Record<string, any>>(config: T): T {
return config
}
const config = createConfig({
apiUrl: 'https://api.example.com',
timeout: 5000
} as const)
// config.apiUrl is 'https://api.example.com' (literal type!)TypeScript 5.4 Features
NoInfer Utility Type
// Prevent type inference from specific positions
function createStore<T>(
initial: T,
merge: (a: T, b: NoInfer<T>) => T
): T {
return merge(initial, initial)
}
const store = createStore(
{ count: 0 },
(a, b) => ({ count: a.count + b.count })
)
// b is inferred from first parameter, not from this lambda!
// Real-world example: React setState
type SetState<T> = (value: T | ((prev: T) => NoInfer<T>)) => voidImport Attributes
// Import JSON with type assertions
import config from './config.json' with { type: 'json' }
// Works with dynamic imports too
const data = await import('./data.json', {
with: { type: 'json' }
})TypeScript 5.3 Features
Import Types Syntax
// Import only types (guaranteed to be erased)
import type { User } from './user'
// Import type and value separately
import { type User, createUser } from './user'
// Resolution mode for .cts/.mts files
import type { RequestHandler } from 'express' with { 'resolution-mode': 'require' }TypeScript 5.0 Features
Decorators (Stage 3)
// Enable in tsconfig.json
{
"compilerOptions": {
"experimentalDecorators": false, // Use standard decorators
}
}
// Class decorator
function logged<T extends { new (...args: any[]): {} }>(constructor: T) {
return class extends constructor {
constructor(...args: any[]) {
console.log(`Creating ${constructor.name}`)
super(...args)
}
}
}
@logged
class User {
constructor(public name: string) {}
}
// Method decorator
function measure(
target: any,
propertyKey: string,
descriptor: PropertyDescriptor
) {
const original = descriptor.value
descriptor.value = async function (...args: any[]) {
const start = performance.now()
const result = await original.apply(this, args)
const duration = performance.now() - start
console.log(`${propertyKey} took ${duration}ms`)
return result
}
return descriptor
}
class API {
@measure
async fetchData() {
await new Promise(resolve => setTimeout(resolve, 100))
return { data: 'result' }
}
}Const Type Parameters (5.0+)
// Make generic type parameters const
type Const<T> = T extends readonly any[] ? readonly [...T] : T
function tuple<const T extends readonly any[]>(...args: T): T {
return args
}
const numbers = tuple(1, 2, 3)
// ^? readonly [1, 2, 3]
// Not: (number | number | number)[]TypeScript 4.9 Features
Satisfies Operator
// Ensure type without widening
type Config = {
url: string
timeout: number
retries?: number
}
// ✅ GOOD: Keeps literal types
const config = {
url: 'https://api.example.com',
timeout: 5000,
retries: 3
} satisfies Config
config.url // 'https://api.example.com' (literal type!)
// ❌ BAD: Widens to string
const config2: Config = {
url: 'https://api.example.com',
timeout: 5000
}
config2.url // string (widened)
// Real-world example: API routes
type Route = {
method: 'GET' | 'POST' | 'PUT' | 'DELETE'
path: string
handler: (req: any) => any
}
const routes = {
getUser: {
method: 'GET',
path: '/users/:id',
handler: (req) => ({ user: 'data' })
},
createUser: {
method: 'POST',
path: '/users',
handler: (req) => ({ created: true })
}
} satisfies Record<string, Route>
routes.getUser.method // 'GET' (literal!)Auto-Accessors in Classes
class User {
accessor name: string = ''
// Equivalent to:
// #__name: string = ''
// get name() { return this.#__name }
// set name(value: string) { this.#__name = value }
}
// With decorators
function logged(target: any, context: ClassAccessorDecoratorContext) {
return {
get(this: any) {
const value = target.get.call(this)
console.log(`Getting ${String(context.name)}: ${value}`)
return value
},
set(this: any, value: any) {
console.log(`Setting ${String(context.name)}: ${value}`)
target.set.call(this, value)
}
}
}
class User {
@logged
accessor name: string = ''
}Advanced Type Patterns
Branded Types
// Prevent mixing similar primitive types
type UserId = string & { readonly __brand: 'UserId' }
type PostId = string & { readonly __brand: 'PostId' }
function UserId(id: string): UserId {
return id as UserId
}
function PostId(id: string): PostId {
return id as PostId
}
function getUser(id: UserId): User { /* ... */ }
function getPost(id: PostId): Post { /* ... */ }
const userId = UserId('user-123')
const postId = PostId('post-456')
getUser(userId) // ✅ OK
getUser(postId) // ❌ Error: PostId not assignable to UserIdTemplate Literal Types
// Type-safe event system
type EventName = `on${Capitalize<string>}`
type Handler<T extends EventName> =
T extends `on${infer Event}`
? (event: Lowercase<Event>) => void
: never
const handlers: Record<EventName, Handler<EventName>> = {
onClick: (event) => console.log(event), // event: 'click'
onMouseMove: (event) => {}, // event: 'mousemove'
}
// Type-safe API paths
type HttpMethod = 'GET' | 'POST' | 'PUT' | 'DELETE'
type ApiPath = `/api/${string}`
type Endpoint = `${HttpMethod} ${ApiPath}`
function registerEndpoint(endpoint: Endpoint, handler: Function) {}
registerEndpoint('GET /api/users', () => {}) // ✅ OK
registerEndpoint('GET api/users', () => {}) // ❌ Error: missing /
registerEndpoint('PATCH /api/users', () => {}) // ❌ Error: invalid methodRecursive Conditional Types
// Deep readonly
type DeepReadonly<T> = {
readonly [K in keyof T]: T[K] extends object
? DeepReadonly<T[K]>
: T[K]
}
interface Config {
db: {
host: string
port: number
credentials: {
username: string
password: string
}
}
}
const config: DeepReadonly<Config> = {
db: {
host: 'localhost',
port: 5432,
credentials: {
username: 'admin',
password: 'secret'
}
}
}
config.db.credentials.password = 'new' // ❌ Error: readonly
// Deep partial
type DeepPartial<T> = {
[K in keyof T]?: T[K] extends object
? DeepPartial<T[K]>
: T[K]
}
type ConfigUpdate = DeepPartial<Config>
const update: ConfigUpdate = {
db: {
port: 5433 // Can update just port
}
}Variadic Tuple Types
// Type-safe function composition
type Func<Args extends any[], Return> = (...args: Args) => Return
function compose<A extends any[], B, C>(
f: Func<[B], C>,
g: Func<A, B>
): Func<A, C> {
return (...args: A) => f(g(...args))
}
const add = (a: number, b: number) => a + b
const double = (n: number) => n * 2
const addThenDouble = compose(double, add)
const result = addThenDouble(2, 3) // 10
// ^? number
// Type-safe tuple concatenation
type Concat<T extends any[], U extends any[]> = [...T, ...U]
type Result = Concat<[1, 2], [3, 4]>
// ^? [1, 2, 3, 4]Strict Mode Configuration
// tsconfig.json
{
"compilerOptions": {
// Strict mode (enables all below)
"strict": true,
// Individual strict flags
"strictNullChecks": true, // null/undefined checking
"strictFunctionTypes": true, // Function parameter checking
"strictBindCallApply": true, // Accurate bind/call/apply
"strictPropertyInitialization": true, // Class property init
"noImplicitAny": true, // No implicit any types
"noImplicitThis": true, // No implicit this
"alwaysStrict": true, // Emit 'use strict'
// Additional safety
"noUncheckedIndexedAccess": true, // obj[key] includes undefined
"noImplicitReturns": true, // All code paths must return
"noFallthroughCasesInSwitch": true, // Switch case fallthrough
"noUnusedLocals": true, // Catch unused variables
"noUnusedParameters": true, // Catch unused params
"noPropertyAccessFromIndexSignature": true, // Force bracket notation
"exactOptionalPropertyTypes": true, // undefined !== missing property
// Module resolution
"moduleResolution": "bundler", // Modern bundler resolution
"allowImportingTsExtensions": true, // Import .ts files
"resolveJsonModule": true, // Import JSON
"isolatedModules": true, // Each file is a module
// Emit
"declaration": true, // Generate .d.ts files
"declarationMap": true, // Source maps for .d.ts
"sourceMap": true, // Generate source maps
"removeComments": false, // Keep comments in output
// Advanced
"skipLibCheck": true, // Skip type checking .d.ts files
"forceConsistentCasingInFileNames": true, // Case-sensitive imports
"useDefineForClassFields": true // ECMAScript-compliant class fields
}
}Best Practices
1. Enable strict mode - Catch bugs at compile time 2. Use `const` type parameters - Preserve literal types 3. Use `satisfies` - Type-check without widening 4. Use branded types - Prevent primitive type confusion 5. Use template literals - Type-safe string patterns 6. Use `NoInfer` - Control type inference direction 7. Prefer `unknown` over `any` - Force type checking 8. Use type predicates - Better type narrowing 9. Enable `noUncheckedIndexedAccess` - Safer array/object access 10. Use decorators - Clean metadata and cross-cutting concerns
Advanced TypeScript Patterns
Exhaustive type checking, branded types, and type guards for production-grade type safety.
Exhaustive Type Checking
TypeScript's type system can guarantee compile-time exhaustiveness for union types. This prevents runtime bugs when union members are added or changed.
The assertNever Pattern
// ALWAYS use this helper function
function assertNever(x: never): never {
throw new Error("Unexpected value: " + x)
}
// Example: Status handling
type AnalysisStatus = 'pending' | 'running' | 'completed' | 'failed'
function getStatusColor(status: AnalysisStatus): string {
switch (status) {
case 'pending': return 'gray'
case 'running': return 'blue'
case 'completed': return 'green'
case 'failed': return 'red'
default: return assertNever(status) // Compile-time exhaustiveness check
}
}
// If you add a new status 'cancelled', TypeScript will error at compile time:
// Error: Argument of type 'string' is not assignable to parameter of type 'never'.Exhaustive Record Mapping
// For mapping union types to values, use satisfies with Record
type EventType = 'click' | 'scroll' | 'keypress' | 'hover'
const eventColors = {
click: 'red',
scroll: 'blue',
keypress: 'green',
hover: 'yellow',
} as const satisfies Record<EventType, string>
// TypeScript will error if any EventType is missing from the record
// Adding new EventType requires updating this recordExhaustive Handler Objects
// For complex logic, use handler objects instead of switches
type ContentType = 'article' | 'video' | 'podcast' | 'repository'
interface ContentHandler<T> {
article: (data: ArticleData) => T
video: (data: VideoData) => T
podcast: (data: PodcastData) => T
repository: (data: RepoData) => T
}
function createContentHandlers<T>(handlers: ContentHandler<T>): ContentHandler<T> {
return handlers
}
// Usage: TypeScript enforces all content types are handled
const renderContent = createContentHandlers({
article: (data) => <ArticleCard {...data} />,
video: (data) => <VideoPlayer {...data} />,
podcast: (data) => <AudioPlayer {...data} />,
repository: (data) => <RepoCard {...data} />,
})Exhaustive Union Checks with Type Guards
// When you need runtime type narrowing with exhaustiveness
type APIResponse =
| { type: 'success'; data: Data }
| { type: 'error'; error: Error }
| { type: 'loading' }
function handleResponse(response: APIResponse): string {
switch (response.type) {
case 'success':
return "Data: " + response.data.id
case 'error':
return "Error: " + response.error.message
case 'loading':
return 'Loading...'
default:
return assertNever(response) // Ensures all cases handled
}
}Template Literal Exhaustiveness
// For string pattern unions
type Size = 'sm' | 'md' | 'lg' | 'xl'
type Variant = 'primary' | 'secondary' | 'danger'
// Exhaustive size mapping
const sizeMap = {
sm: 'text-sm py-1 px-2',
md: 'text-base py-2 px-4',
lg: 'text-lg py-3 px-6',
xl: 'text-xl py-4 px-8',
} as const satisfies Record<Size, string>
// Compile-time error if Size is expanded without updating sizeMapBranded Types
Prevent mixing similar primitive types at compile time.
TypeScript Pattern (with Zod Runtime Validation)
import { z } from 'zod'
// Create branded types for different ID kinds
const UserId = z.string().uuid().brand<'UserId'>()
const AnalysisId = z.string().uuid().brand<'AnalysisId'>()
const ArtifactId = z.string().uuid().brand<'ArtifactId'>()
type UserId = z.infer<typeof UserId>
type AnalysisId = z.infer<typeof AnalysisId>
type ArtifactId = z.infer<typeof ArtifactId>
// Now TypeScript prevents mixing ID types
function deleteAnalysis(id: AnalysisId): void { ... }
function getUser(id: UserId): User { ... }
const userId: UserId = UserId.parse('...')
const analysisId: AnalysisId = AnalysisId.parse('...')
deleteAnalysis(analysisId) // OK
deleteAnalysis(userId) // Error: UserId not assignable to AnalysisIdPython Pattern (NewType Compile-Time Safety)
from typing import NewType
from uuid import UUID
# Define branded types (zero runtime overhead)
AnalysisID = NewType("AnalysisID", UUID)
ArtifactID = NewType("ArtifactID", UUID)
SessionID = NewType("SessionID", UUID)
TraceID = NewType("TraceID", str)
# Factory functions for runtime validation
def create_analysis_id(value: UUID | str) -> AnalysisID:
"""Create typed AnalysisID with validation."""
if isinstance(value, str):
value = UUID(value)
return AnalysisID(value)
def create_artifact_id(value: UUID | str) -> ArtifactID:
"""Create typed ArtifactID with validation."""
if isinstance(value, str):
value = UUID(value)
return ArtifactID(value)
# Type checker (mypy/ty) prevents mixing
def delete_analysis(id: AnalysisID) -> None: ...
def get_artifact(id: ArtifactID) -> Artifact: ...
analysis_id = create_analysis_id("...")
artifact_id = create_artifact_id("...")
delete_analysis(analysis_id) # OK
delete_analysis(artifact_id) # Error: ArtifactID not assignable to AnalysisIDWhy NewType for Python?
- Zero runtime overhead - compiled away, no wrapper object
- Mypy/Ty enforcement - catches ID mixing at type-check time
- Explicit factories - centralized validation logic
- Better than Pydantic for this use case - no serialization needed
Pure TypeScript Branded Types (No Runtime Library)
// Prevent mixing similar primitive types
type UserId = string & { readonly __brand: 'UserId' }
type PostId = string & { readonly __brand: 'PostId' }
function UserId(id: string): UserId {
return id as UserId
}
function PostId(id: string): PostId {
return id as PostId
}
function getUser(id: UserId): User { /* ... */ }
function getPost(id: PostId): Post { /* ... */ }
const userId = UserId('user-123')
const postId = PostId('post-456')
getUser(userId) // OK
getUser(postId) // Error: PostId not assignable to UserIdType Guards
Custom type narrowing functions for complex runtime checks.
Basic Type Guards
// Type guard with type predicate
function isString(value: unknown): value is string {
return typeof value === 'string'
}
function isUser(obj: unknown): obj is User {
return (
typeof obj === 'object' &&
obj !== null &&
'id' in obj &&
'email' in obj &&
typeof (obj as User).id === 'string' &&
typeof (obj as User).email === 'string'
)
}
// Usage
const data: unknown = await fetchData()
if (isUser(data)) {
console.log(data.email) // TypeScript knows data is User
}Assertion Functions
// Assertion function throws on failure
function assertIsUser(obj: unknown): asserts obj is User {
if (!isUser(obj)) {
throw new Error('Expected User object')
}
}
// Usage - narrows type after call
const data: unknown = await fetchData()
assertIsUser(data)
console.log(data.email) // TypeScript knows data is UserDiscriminated Union Guards
type Result<T, E> =
| { success: true; value: T }
| { success: false; error: E }
function isSuccess<T, E>(result: Result<T, E>): result is { success: true; value: T } {
return result.success === true
}
function isError<T, E>(result: Result<T, E>): result is { success: false; error: E } {
return result.success === false
}
// Usage
const result = await doSomething()
if (isSuccess(result)) {
console.log(result.value) // T
} else {
console.log(result.error) // E
}Common Anti-Patterns
// NEVER use non-exhaustive switch
switch (status) {
case 'pending': return 'gray'
case 'running': return 'blue'
// Missing cases! Runtime bugs waiting to happen
}
// NEVER use default without assertNever
switch (status) {
case 'pending': return 'gray'
case 'running': return 'blue'
default: return 'unknown' // Silent bug if new status added
}
// NEVER use if-else chains for union types
if (status === 'pending') return 'gray'
else if (status === 'running') return 'blue'
// No compile-time check for missing cases!
// ALWAYS use switch with assertNever
switch (status) {
case 'pending': return 'gray'
case 'running': return 'blue'
case 'completed': return 'green'
case 'failed': return 'red'
default: return assertNever(status)
}Best Practices
1. Exhaustive switches - Always use assertNever in default case 2. Exhaustive records - Use satisfies Record<UnionType, Value> 3. Branded types (TypeScript) - Use Zod .brand<>() for distinct ID types 4. Branded types (Python) - Use NewType for zero-overhead compile-time safety 5. Type guards - Create reusable predicates for complex type narrowing 6. Assertion functions - Use asserts for imperative type narrowing
Zod Schema Patterns
Complete guide to Zod runtime validation patterns for TypeScript applications.
Core Schema Definition
Basic Types
import { z } from 'zod'
// Primitives
const StringSchema = z.string()
const NumberSchema = z.number()
const BooleanSchema = z.boolean()
const DateSchema = z.date()
const BigIntSchema = z.bigint()
const UndefinedSchema = z.undefined()
const NullSchema = z.null()
const AnySchema = z.any()
const UnknownSchema = z.unknown()
const NeverSchema = z.never()
const VoidSchema = z.void()
// Special types
const LiteralSchema = z.literal('admin') // Only accepts 'admin'
const EnumSchema = z.enum(['admin', 'user', 'guest'])
const NativeEnumSchema = z.nativeEnum(UserRole) // From TypeScript enumObject Schemas
// Basic object
const UserSchema = z.object({
id: z.string().uuid(),
email: z.string().email(),
name: z.string().min(1).max(100),
age: z.number().int().positive().max(120),
role: z.enum(['admin', 'user']),
isActive: z.boolean().default(true),
metadata: z.record(z.string()).optional(),
createdAt: z.date().default(() => new Date())
})
// Infer TypeScript type
type User = z.infer<typeof UserSchema>
// {
// id: string;
// email: string;
// name: string;
// age: number;
// role: 'admin' | 'user';
// isActive: boolean;
// metadata?: Record<string, string>;
// createdAt: Date;
// }
// Partial, pick, omit
const PartialUserSchema = UserSchema.partial() // All fields optional
const UpdateUserSchema = UserSchema.pick({ name: true, email: true })
const PublicUserSchema = UserSchema.omit({ metadata: true })
// Extend schemas
const AdminSchema = UserSchema.extend({
permissions: z.array(z.string()),
department: z.string()
})
// Merge schemas
const TimestampsSchema = z.object({
createdAt: z.date(),
updatedAt: z.date()
})
const UserWithTimestamps = UserSchema.merge(TimestampsSchema)Array and Tuple Schemas
// Arrays
const StringArraySchema = z.array(z.string())
const NumberArraySchema = z.number().array() // Alternative syntax
const UserArraySchema = z.array(UserSchema).min(1).max(100)
// Non-empty arrays
const TagsSchema = z.array(z.string()).nonempty()
// Tuples (fixed-length arrays)
const CoordinateSchema = z.tuple([z.number(), z.number()])
// type Coordinate = [number, number]
const ResponseSchema = z.tuple([
z.number(), // status code
z.string(), // message
z.unknown() // data
])
// Rest parameters
const VariadicTupleSchema = z.tuple([z.string()]).rest(z.number())
// [string, ...number[]]Transformations and Refinements
Basic Transformations
// Transform data after validation
const EmailSchema = z.string()
.email()
.transform(email => email.toLowerCase().trim())
const TimestampSchema = z.string()
.transform(str => new Date(str))
const PriceSchema = z.number()
.transform(cents => cents / 100) // Store cents, return dollars
// Chained transformations
const SlugSchema = z.string()
.transform(str => str.toLowerCase())
.transform(str => str.replace(/\s+/g, '-'))
.transform(str => str.replace(/[^a-z0-9-]/g, ''))
// Transform with error handling
const JSONSchema = z.string().transform((str, ctx) => {
try {
return JSON.parse(str)
} catch (e) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: 'Invalid JSON'
})
return z.NEVER
}
})Refinements (Custom Validation)
// Single refinement
const PasswordSchema = z.string()
.min(8)
.refine((pass) => /[A-Z]/.test(pass), {
message: 'Password must contain at least one uppercase letter'
})
.refine((pass) => /[a-z]/.test(pass), {
message: 'Password must contain at least one lowercase letter'
})
.refine((pass) => /[0-9]/.test(pass), {
message: 'Password must contain at least one number'
})
.refine((pass) => /[^A-Za-z0-9]/.test(pass), {
message: 'Password must contain at least one special character'
})
// Multiple field refinement
const DateRangeSchema = z.object({
startDate: z.date(),
endDate: z.date()
}).refine(data => data.endDate > data.startDate, {
message: 'End date must be after start date',
path: ['endDate'] // Which field to attach error to
})
// Async refinement (e.g., unique email check)
const UniqueEmailSchema = z.string()
.email()
.refine(async (email) => {
const exists = await db.user.findUnique({ where: { email } })
return !exists
}, {
message: 'Email already exists'
})
// Superrefine for complex validation
const PaymentSchema = z.object({
method: z.enum(['card', 'paypal', 'bank']),
cardNumber: z.string().optional(),
paypalEmail: z.string().optional(),
bankAccount: z.string().optional()
}).superRefine((data, ctx) => {
if (data.method === 'card' && !data.cardNumber) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: 'Card number is required',
path: ['cardNumber']
})
}
if (data.method === 'paypal' && !data.paypalEmail) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: 'PayPal email is required',
path: ['paypalEmail']
})
}
})Union and Discriminated Union Patterns
Basic Unions
// Simple union
const StringOrNumberSchema = z.union([z.string(), z.number()])
// Nullable/Optional
const NullableStringSchema = z.string().nullable()
const OptionalStringSchema = z.string().optional()
const NullishStringSchema = z.string().nullish() // null | undefined
// Multiple types
const IdSchema = z.union([
z.string().uuid(),
z.number().int().positive()
])Discriminated Unions (Recommended)
// Event types with discriminator
const EventSchema = z.discriminatedUnion('type', [
z.object({
type: z.literal('click'),
x: z.number(),
y: z.number(),
button: z.enum(['left', 'right', 'middle'])
}),
z.object({
type: z.literal('scroll'),
offset: z.number(),
direction: z.enum(['up', 'down'])
}),
z.object({
type: z.literal('resize'),
width: z.number(),
height: z.number()
})
])
type Event = z.infer<typeof EventSchema>
// Event will be a proper discriminated union
// API Response pattern
const ApiResponseSchema = z.discriminatedUnion('status', [
z.object({
status: z.literal('success'),
data: z.unknown()
}),
z.object({
status: z.literal('error'),
error: z.object({
code: z.string(),
message: z.string()
})
})
])Recursive and Lazy Schemas
// Recursive type (e.g., nested categories)
interface Category {
id: string
name: string
children?: Category[]
}
const CategorySchema: z.ZodType<Category> = z.lazy(() =>
z.object({
id: z.string(),
name: z.string(),
children: z.array(CategorySchema).optional()
})
)
// File system tree
interface FileNode {
name: string
type: 'file' | 'directory'
children?: FileNode[]
}
const FileNodeSchema: z.ZodType<FileNode> = z.lazy(() =>
z.object({
name: z.string(),
type: z.enum(['file', 'directory']),
children: z.array(FileNodeSchema).optional()
})
)Error Handling
Parse vs SafeParse
// .parse() - throws on error (use when you're confident)
try {
const user = UserSchema.parse(data)
// user is typed as User
} catch (error) {
if (error instanceof z.ZodError) {
console.error(error.issues)
}
}
// .safeParse() - returns result object (recommended)
const result = UserSchema.safeParse(data)
if (result.success) {
const user = result.data
// user is typed as User
} else {
const errors = result.error.issues
// Handle validation errors
}Custom Error Messages
const schema = z.object({
email: z.string({
required_error: 'Email is required',
invalid_type_error: 'Email must be a string'
}).email({ message: 'Invalid email format' }),
age: z.number({
required_error: 'Age is required',
invalid_type_error: 'Age must be a number'
}).min(18, { message: 'Must be at least 18 years old' })
.max(120, { message: 'Age seems unrealistic' })
})
// Custom error map for all validations
z.setErrorMap((issue, ctx) => {
if (issue.code === z.ZodIssueCode.invalid_type) {
if (issue.expected === 'string') {
return { message: 'Bad type!' }
}
}
return { message: ctx.defaultError }
})Formatting Errors for Users
function formatZodErrors(error: z.ZodError): Record<string, string> {
return error.issues.reduce((acc, issue) => {
const path = issue.path.join('.')
acc[path] = issue.message
return acc
}, {} as Record<string, string>)
}
const result = UserSchema.safeParse(data)
if (!result.success) {
const fieldErrors = formatZodErrors(result.error)
// { "email": "Invalid email format", "age": "Must be at least 18" }
}Best Practices
Schema Organization
// ✅ GOOD: Reusable schemas
const EmailSchema = z.string().email().toLowerCase()
const UuidSchema = z.string().uuid()
const TimestampSchema = z.date().default(() => new Date())
const UserSchema = z.object({
id: UuidSchema,
email: EmailSchema,
createdAt: TimestampSchema
})
// ❌ BAD: Inline schemas (hard to reuse)
const UserSchema = z.object({
email: z.string().email().transform(e => e.toLowerCase()),
// ... duplicated in every schema
})Performance
// ✅ Define schemas once (outside functions/components)
const UserSchema = z.object({ /* ... */ })
export function validateUser(data: unknown) {
return UserSchema.safeParse(data)
}
// ❌ Don't create schemas in hot paths
export function validateUser(data: unknown) {
const schema = z.object({ /* ... */ }) // Created every call!
return schema.safeParse(data)
}Type Inference
// ✅ GOOD: Infer types from schemas (single source of truth)
const UserSchema = z.object({
id: z.string(),
name: z.string()
})
type User = z.infer<typeof UserSchema>
// ❌ BAD: Separate type and schema (can drift apart)
interface User {
id: string
name: string
}
const UserSchema = z.object({
id: z.string(),
name: z.string()
})Comparison with Pydantic (Python)
For Python developers familiar with Pydantic:
# Pydantic (Python)
from pydantic import BaseModel, EmailStr, validator
class User(BaseModel):
id: str
email: EmailStr
age: int
@validator('age')
def validate_age(cls, v):
if v < 18:
raise ValueError('Must be 18+')
return v// Zod (TypeScript) - equivalent
const UserSchema = z.object({
id: z.string(),
email: z.string().email(),
age: z.number().min(18, { message: 'Must be 18+' })
})
type User = z.infer<typeof UserSchema>Key differences:
- Zod uses method chaining, Pydantic uses decorators
- Zod types are inferred, Pydantic types are declared
- Both provide excellent runtime validation
- Both support custom validators/refinements
/**
* tRPC Router Template
*
* Production-ready tRPC router with authentication, validation, and error handling.
*/
import { initTRPC, TRPCError } from '@trpc/server'
import { z } from 'zod'
import type { CreateNextContextOptions } from '@trpc/server/adapters/next'
import superjson from 'superjson'
// ====================
// Context Definition
// ====================
export interface Context {
userId?: string
sessionId?: string
// db: PrismaClient
// redis: Redis
req: Request
}
export async function createContext(
opts: CreateNextContextOptions
): Promise<Context> {
// Extract auth token
const token = opts.req.headers.get('authorization')?.replace('Bearer ', '')
// Verify token and get user ID
const userId = token ? await verifyJWT(token) : undefined
const sessionId = token ? await getSessionId(token) : undefined
return {
userId,
sessionId,
req: opts.req,
}
}
// ====================
// tRPC Initialization
// ====================
const t = initTRPC.context<Context>().create({
transformer: superjson,
errorFormatter({ shape, error }) {
return {
...shape,
data: {
...shape.data,
zodError:
error.cause instanceof z.ZodError
? error.cause.flatten()
: null,
},
}
},
})
export const router = t.router
export const publicProcedure = t.procedure
export const middleware = t.middleware
// ====================
// Middleware
// ====================
// Authentication middleware
const enforceAuth = middleware(async ({ ctx, next }) => {
if (!ctx.userId) {
throw new TRPCError({
code: 'UNAUTHORIZED',
message: 'You must be logged in',
})
}
return next({
ctx: {
...ctx,
userId: ctx.userId, // Now guaranteed non-null
},
})
})
// Rate limiting middleware
const rateLimit = (limit: number, windowMs: number) =>
middleware(async ({ ctx, next, path }) => {
const key = `ratelimit:${ctx.userId || 'anon'}:${path}`
// Implement rate limiting logic
// const count = await redis.incr(key)
// if (count === 1) await redis.expire(key, windowMs / 1000)
// if (count > limit) throw new TRPCError({ code: 'TOO_MANY_REQUESTS' })
return next()
})
// Logging middleware
const logger = middleware(async ({ ctx, next, path, type }) => {
const start = Date.now()
console.log(`→ ${type} ${path}`, {
userId: ctx.userId,
sessionId: ctx.sessionId,
})
const result = await next()
const duration = Date.now() - start
console.log(`← ${type} ${path} - ${duration}ms`)
return result
})
// ====================
// Procedures
// ====================
export const protectedProcedure = publicProcedure
.use(logger)
.use(enforceAuth)
export const rateLimitedProcedure = publicProcedure
.use(logger)
.use(rateLimit(100, 60000)) // 100 requests per minute
export const adminProcedure = protectedProcedure
.use(middleware(async ({ ctx, next }) => {
// Check if user is admin
// const user = await db.user.findUnique({ where: { id: ctx.userId } })
// if (user?.role !== 'admin') {
// throw new TRPCError({ code: 'FORBIDDEN' })
// }
return next()
}))
// ====================
// Example Router: Posts
// ====================
const PostInputSchema = z.object({
title: z.string().min(1).max(200),
content: z.string().min(1),
published: z.boolean().default(false),
tags: z.array(z.string()).max(10).default([]),
})
export const postRouter = router({
// List posts with cursor pagination
list: publicProcedure
.input(z.object({
limit: z.number().min(1).max(100).default(10),
cursor: z.string().optional(),
published: z.boolean().optional(),
authorId: z.string().optional(),
}))
.query(async ({ ctx, input }) => {
// Fetch posts from database
// const posts = await db.post.findMany({
// take: input.limit + 1,
// cursor: input.cursor ? { id: input.cursor } : undefined,
// where: {
// published: input.published,
// authorId: input.authorId,
// },
// orderBy: { createdAt: 'desc' },
// include: { author: true },
// })
// Mock data
const posts: any[] = []
let nextCursor: string | undefined
if (posts.length > input.limit) {
const nextItem = posts.pop()
nextCursor = nextItem!.id
}
return {
items: posts,
nextCursor,
}
}),
// Infinite query variant
infinite: publicProcedure
.input(z.object({
limit: z.number().min(1).max(50).default(10),
cursor: z.string().optional(),
}))
.query(async ({ input }) => {
// Same as list, but optimized for infinite scroll
const posts: any[] = []
return {
items: posts,
nextCursor: posts.length > 0 ? posts[posts.length - 1].id : undefined,
}
}),
// Get single post by ID
byId: publicProcedure
.input(z.object({ id: z.string() }))
.query(async ({ input }) => {
// const post = await db.post.findUnique({
// where: { id: input.id },
// include: { author: true, comments: true },
// })
const post = null
if (!post) {
throw new TRPCError({
code: 'NOT_FOUND',
message: `Post with ID ${input.id} not found`,
})
}
return post
}),
// Get multiple posts by IDs (batching example)
byIds: publicProcedure
.input(z.object({
ids: z.array(z.string()).min(1).max(100),
}))
.query(async ({ input }) => {
// const posts = await db.post.findMany({
// where: { id: { in: input.ids } },
// })
return []
}),
// Create post (protected)
create: protectedProcedure
.input(PostInputSchema)
.mutation(async ({ ctx, input }) => {
// Create post in database
// const post = await db.post.create({
// data: {
// ...input,
// authorId: ctx.userId,
// },
// })
const post = { id: 'mock-id', ...input, authorId: ctx.userId }
return post
}),
// Update post (protected)
update: protectedProcedure
.input(z.object({
id: z.string(),
data: PostInputSchema.partial(),
}))
.mutation(async ({ ctx, input }) => {
// Check ownership
// const existing = await db.post.findUnique({
// where: { id: input.id },
// })
const existing = null
if (!existing) {
throw new TRPCError({
code: 'NOT_FOUND',
message: 'Post not found',
})
}
// if (existing.authorId !== ctx.userId) {
// throw new TRPCError({
// code: 'FORBIDDEN',
// message: 'You can only edit your own posts',
// })
// }
// Update post
// const updated = await db.post.update({
// where: { id: input.id },
// data: input.data,
// })
return existing
}),
// Delete post (protected)
delete: protectedProcedure
.input(z.object({ id: z.string() }))
.mutation(async ({ ctx, input }) => {
// Check ownership
// const post = await db.post.findUnique({
// where: { id: input.id },
// })
const post = null
if (!post) {
throw new TRPCError({ code: 'NOT_FOUND' })
}
// if (post.authorId !== ctx.userId) {
// throw new TRPCError({ code: 'FORBIDDEN' })
// }
// Delete post
// await db.post.delete({ where: { id: input.id } })
return { success: true }
}),
// Publish/unpublish post
publish: protectedProcedure
.input(z.object({
id: z.string(),
published: z.boolean(),
}))
.mutation(async ({ ctx, input }) => {
// Check ownership and update
// const post = await db.post.update({
// where: {
// id: input.id,
// authorId: ctx.userId, // Only author can publish
// },
// data: { published: input.published },
// })
return { success: true }
}),
})
// ====================
// Example Router: Users
// ====================
export const userRouter = router({
// Get current user
me: protectedProcedure
.query(async ({ ctx }) => {
// const user = await db.user.findUnique({
// where: { id: ctx.userId },
// select: {
// id: true,
// email: true,
// name: true,
// role: true,
// },
// })
return null
}),
// Update current user
update: protectedProcedure
.input(z.object({
name: z.string().min(1).max(100).optional(),
bio: z.string().max(500).optional(),
}))
.mutation(async ({ ctx, input }) => {
// const updated = await db.user.update({
// where: { id: ctx.userId },
// data: input,
// })
return { success: true }
}),
// Get user by ID (public profile)
byId: publicProcedure
.input(z.object({ id: z.string() }))
.query(async ({ input }) => {
// const user = await db.user.findUnique({
// where: { id: input.id },
// select: {
// id: true,
// name: true,
// bio: true,
// avatar: true,
// },
// })
return null
}),
})
// ====================
// Root Router
// ====================
export const appRouter = router({
post: postRouter,
user: userRouter,
// Health check
health: publicProcedure
.query(() => ({
status: 'ok',
timestamp: new Date(),
})),
})
export type AppRouter = typeof appRouter
// ====================
// Helper Functions
// ====================
async function verifyJWT(token: string): Promise<string | undefined> {
// Implement JWT verification
// const payload = await jwt.verify(token, JWT_SECRET)
// return payload.userId
return undefined
}
async function getSessionId(token: string): Promise<string | undefined> {
// Extract session ID from token
return undefined
}
// ====================
// Server-Side Caller
// ====================
export const createCaller = t.createCallerFactory(appRouter)
// Usage in server components:
// const caller = createCaller(await createContext(...))
// const posts = await caller.post.list({ limit: 10 })
/**
* Common Zod Schema Patterns
*
* Reusable validation schemas for typical application needs.
*/
import { z } from 'zod'
// ====================
// Common Base Schemas
// ====================
export const EmailSchema = z.string().email().toLowerCase().trim()
export const PasswordSchema = z.string()
.min(8, 'Password must be at least 8 characters')
.max(100)
.refine((pass) => /[A-Z]/.test(pass), 'Must contain uppercase letter')
.refine((pass) => /[a-z]/.test(pass), 'Must contain lowercase letter')
.refine((pass) => /[0-9]/.test(pass), 'Must contain number')
.refine((pass) => /[^A-Za-z0-9]/.test(pass), 'Must contain special character')
export const UuidSchema = z.string().uuid()
export const UrlSchema = z.string().url()
export const DateStringSchema = z.string().datetime().transform(str => new Date(str))
export const PhoneSchema = z.string().regex(
/^\+?[1-9]\d{1,14}$/,
'Invalid phone number format (E.164)'
)
export const SlugSchema = z.string()
.min(1)
.max(100)
.regex(/^[a-z0-9]+(?:-[a-z0-9]+)*$/, 'Must be lowercase with hyphens only')
// ====================
// Pagination Schemas
// ====================
export const CursorPaginationSchema = z.object({
limit: z.number().int().min(1).max(100).default(10),
cursor: z.string().optional(),
})
export const OffsetPaginationSchema = z.object({
limit: z.number().int().min(1).max(100).default(10),
offset: z.number().int().min(0).default(0),
})
export const PagePaginationSchema = z.object({
page: z.number().int().min(1).default(1),
pageSize: z.number().int().min(1).max(100).default(10),
})
// ====================
// User Schemas
// ====================
export const UserCreateSchema = z.object({
email: EmailSchema,
password: PasswordSchema,
name: z.string().min(1).max(100),
role: z.enum(['admin', 'user', 'guest']).default('user'),
})
export const UserUpdateSchema = UserCreateSchema.partial().omit({ password: true })
export const UserLoginSchema = z.object({
email: EmailSchema,
password: z.string().min(1),
})
export const UserResponseSchema = z.object({
id: UuidSchema,
email: EmailSchema,
name: z.string(),
role: z.enum(['admin', 'user', 'guest']),
createdAt: z.date(),
updatedAt: z.date(),
})
export type UserCreate = z.infer<typeof UserCreateSchema>
export type UserUpdate = z.infer<typeof UserUpdateSchema>
export type UserLogin = z.infer<typeof UserLoginSchema>
export type UserResponse = z.infer<typeof UserResponseSchema>
// ====================
// API Request/Response Schemas
// ====================
export const ApiErrorSchema = z.object({
code: z.string(),
message: z.string(),
details: z.record(z.unknown()).optional(),
timestamp: z.date().default(() => new Date()),
})
export const ApiSuccessSchema = <T extends z.ZodTypeAny>(dataSchema: T) =>
z.object({
success: z.literal(true),
data: dataSchema,
meta: z.object({
timestamp: z.date().default(() => new Date()),
}),
})
export const ApiErrorResponseSchema = z.object({
success: z.literal(false),
error: ApiErrorSchema,
})
// Discriminated union for API responses
export const ApiResponseSchema = <T extends z.ZodTypeAny>(dataSchema: T) =>
z.discriminatedUnion('success', [
ApiSuccessSchema(dataSchema),
ApiErrorResponseSchema,
])
// Usage example:
// const UserApiResponse = ApiResponseSchema(UserResponseSchema)
// ====================
// File Upload Schemas
// ====================
export const FileUploadSchema = z.object({
name: z.string().min(1).max(255),
size: z.number().int().positive().max(10 * 1024 * 1024), // 10MB max
type: z.string().regex(/^[a-z]+\/[a-z0-9\-\+\.]+$/i), // MIME type
url: UrlSchema,
})
export const ImageUploadSchema = FileUploadSchema.extend({
type: z.enum([
'image/jpeg',
'image/png',
'image/gif',
'image/webp',
'image/svg+xml'
]),
width: z.number().int().positive().optional(),
height: z.number().int().positive().optional(),
})
// ====================
// Search and Filter Schemas
// ====================
export const SearchSchema = z.object({
query: z.string().min(1).max(100),
filters: z.record(z.union([
z.string(),
z.number(),
z.boolean(),
z.array(z.string()),
])).optional(),
sort: z.object({
field: z.string(),
order: z.enum(['asc', 'desc']),
}).optional(),
pagination: CursorPaginationSchema.or(OffsetPaginationSchema),
})
// ====================
// Date Range Schemas
// ====================
export const DateRangeSchema = z.object({
startDate: z.date(),
endDate: z.date(),
}).refine(
data => data.endDate >= data.startDate,
{
message: 'End date must be after or equal to start date',
path: ['endDate'],
}
)
export const TimestampRangeSchema = z.object({
from: z.number().int().positive(),
to: z.number().int().positive(),
}).refine(
data => data.to >= data.from,
{
message: 'To timestamp must be after or equal to from timestamp',
path: ['to'],
}
)
// ====================
// Address Schemas
// ====================
export const AddressSchema = z.object({
street: z.string().min(1).max(200),
city: z.string().min(1).max(100),
state: z.string().min(2).max(100),
country: z.string().length(2), // ISO 3166-1 alpha-2
postalCode: z.string().min(3).max(10),
coordinates: z.object({
lat: z.number().min(-90).max(90),
lng: z.number().min(-180).max(180),
}).optional(),
})
// ====================
// Nested Object Schemas
// ====================
export const PostCreateSchema = z.object({
title: z.string().min(1).max(200),
content: z.string().min(1).max(10000),
excerpt: z.string().max(500).optional(),
published: z.boolean().default(false),
tags: z.array(z.string().min(1).max(50)).max(10),
metadata: z.object({
readTime: z.number().int().positive().optional(),
featuredImage: ImageUploadSchema.optional(),
}).optional(),
author: z.object({
id: UuidSchema,
name: z.string(),
}),
})
export type PostCreate = z.infer<typeof PostCreateSchema>
// ====================
// Discriminated Union Schemas
// ====================
export const NotificationSchema = z.discriminatedUnion('type', [
z.object({
type: z.literal('email'),
to: EmailSchema,
subject: z.string().min(1).max(200),
body: z.string().min(1),
}),
z.object({
type: z.literal('sms'),
to: PhoneSchema,
message: z.string().min(1).max(160),
}),
z.object({
type: z.literal('push'),
userId: UuidSchema,
title: z.string().min(1).max(100),
body: z.string().min(1).max(200),
}),
])
export type Notification = z.infer<typeof NotificationSchema>
// ====================
// Payment Schemas
// ====================
export const PaymentSchema = z.object({
amount: z.number().positive().multipleOf(0.01), // Cents precision
currency: z.string().length(3).toUpperCase(), // ISO 4217
method: z.enum(['card', 'paypal', 'stripe', 'bank_transfer']),
metadata: z.record(z.string()).optional(),
})
export const CardPaymentSchema = PaymentSchema.extend({
method: z.literal('card'),
cardDetails: z.object({
number: z.string().regex(/^\d{13,19}$/),
expMonth: z.number().int().min(1).max(12),
expYear: z.number().int().min(new Date().getFullYear()),
cvv: z.string().regex(/^\d{3,4}$/),
}),
})
// ====================
// Recursive Schemas
// ====================
type Category = {
id: string
name: string
children?: Category[]
}
export const CategorySchema: z.ZodType<Category> = z.lazy(() =>
z.object({
id: UuidSchema,
name: z.string().min(1).max(100),
children: z.array(CategorySchema).optional(),
})
)
// ====================
// Environment Variable Schemas
// ====================
export const EnvSchema = z.object({
NODE_ENV: z.enum(['development', 'test', 'production']),
DATABASE_URL: z.string().url(),
REDIS_URL: z.string().url().optional(),
JWT_SECRET: z.string().min(32),
API_PORT: z.coerce.number().int().min(1).max(65535).default(3000),
CORS_ORIGIN: z.string().url().or(z.literal('*')),
LOG_LEVEL: z.enum(['debug', 'info', 'warn', 'error']).default('info'),
})
export type Env = z.infer<typeof EnvSchema>
// Validate environment variables at startup
export function validateEnv(): Env {
const result = EnvSchema.safeParse(process.env)
if (!result.success) {
console.error('❌ Invalid environment variables:')
console.error(result.error.flatten().fieldErrors)
process.exit(1)
}
return result.data
}
// ====================
// Form Validation Schemas
// ====================
export const ContactFormSchema = z.object({
name: z.string().min(2).max(100),
email: EmailSchema,
subject: z.string().min(5).max(200),
message: z.string().min(10).max(2000),
consent: z.boolean().refine(val => val === true, {
message: 'You must accept the privacy policy',
}),
})
export type ContactForm = z.infer<typeof ContactFormSchema>
// ====================
// Async Validation Example
// ====================
export const UniqueEmailSchema = z.string()
.email()
.refine(async (email) => {
// Simulate DB check
// const exists = await db.user.findUnique({ where: { email } })
// return !exists
return true
}, {
message: 'Email already exists',
})
// ====================
// Custom Transform Examples
// ====================
export const TrimmedStringSchema = z.string().transform(s => s.trim())
export const CommaSeparatedSchema = z.string()
.transform(s => s.split(',').map(item => item.trim()).filter(Boolean))
export const JsonStringSchema = z.string().transform((str, ctx) => {
try {
return JSON.parse(str)
} catch (e) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: 'Invalid JSON',
})
return z.NEVER
}
})
// ====================
// Error Formatting Utility
// ====================
export function formatZodError(error: z.ZodError): Record<string, string> {
return error.issues.reduce((acc, issue) => {
const path = issue.path.join('.')
acc[path] = issue.message
return acc
}, {} as Record<string, string>)
}
// Usage:
// const result = UserCreateSchema.safeParse(data)
// if (!result.success) {
// const errors = formatZodError(result.error)
// // { "email": "Invalid email", "password": "Must contain uppercase" }
// }