Code Generation Guide¶

Spikard's code generator transforms API schemas into type-safe, production-ready handlers and DTOs for Python, TypeScript, Rust, Ruby, PHP, and Elixir. Generated code integrates with Spikard's runtime and passes strict quality tools (ruff/mypy, tsc, clippy, steep/rubocop, phpstan, mix format).

This guide covers the complete workflow: schema to generated code to integration and validation.

Overview and Workflow¶

The spikard generate command is the entry point for all code generation. It follows a consistent pattern across all protocols:

spikard generate <protocol> <schema-path> --lang <language> [options]

Generation Workflow¶

Schema File
    ↓
Parse & Validate Schema
    ↓
Generate Types/DTOs
    ↓
Generate Handlers/Resolvers
    ↓
Format Code (language-specific)
    ↓
Validate Quality (mypy/tsc/steep/phpstan/clippy)
    ↓
Output Files

Supported Protocols¶

OpenAPI 3.1: REST APIs with request/response validation
AsyncAPI 3.0: WebSocket and Server-Sent Events
GraphQL: Schema-first GraphQL APIs with resolvers
OpenRPC: JSON-RPC 2.0 APIs
Protobuf: gRPC services with binary serialization

Supported Languages¶

All generators support these six target languages:

Python: Type hints (3.10+), dataclasses, mypy --strict compatible
TypeScript: Strict mode, interfaces, biome/eslint compatible
Rust: Strongly-typed structs, cargo clippy clean
Ruby: RBS signatures, steep type checking
PHP: 8.2+, strict types, phpstan level max
Elixir: formatter-clean modules and router scaffolding

OpenAPI Code Generation¶

Generate type-safe REST API handlers from OpenAPI 3.1 specifications.

Basic Usage¶

# Generate Python handlers
spikard generate openapi openapi.yaml --lang python --output ./generated

# Generate TypeScript with custom target
spikard generate openapi openapi.json --lang typescript --output ./src/generated

Example Schema¶

# todo-api.openapi.yaml
openapi: 3.1.0
info:
  title: Todo API
  version: 1.0.0

paths:
  /todos:
    get:
      summary: List todos
      parameters:
        - name: status
          in: query
          schema:
            type: string
            enum: [pending, completed]
      responses:
        '200':
          description: Success
          content:
            application/json:
              schema:
                type: object
                properties:
                  todos:
                    type: array
                    items:
                      $ref: '#/components/schemas/Todo'

    post:
      summary: Create todo
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/CreateTodoRequest'
      responses:
        '201':
          description: Created
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Todo'

components:
  schemas:
    Todo:
      type: object
      required: [id, title, status]
      properties:
        id:
          type: string
          format: uuid
        title:
          type: string
          minLength: 1
          maxLength: 200
        status:
          type: string
          enum: [pending, completed]

    CreateTodoRequest:
      type: object
      required: [title]
      properties:
        title:
          type: string
          minLength: 1
          maxLength: 200

Generated Output (Python)¶

spikard generate openapi todo-api.openapi.yaml --lang python --output ./generated/todo_handlers.py

Generated: generated/todo_handlers.py

#!/usr/bin/env python3
# DO NOT EDIT - Auto-generated by Spikard CLI

from dataclasses import dataclass
from typing import Optional, List
from uuid import UUID
from spikard import Request, Response, Handler

@dataclass
class Todo:
    """A todo item."""
    id: UUID
    title: str
    status: str  # Literal["pending", "completed"]

@dataclass
class CreateTodoRequest:
    """Request to create a todo."""
    title: str

class ListTodosHandler(Handler):
    """Handler for GET /todos."""

    async def handle(self, request: Request) -> Response:
        # Extract query parameters
        status: Optional[str] = request.query.get("status")

        # TODO: Implement handler logic
        todos: List[Todo] = []

        return Response(
            status_code=200,
            body={"todos": [t.__dict__ for t in todos]}
        )

class CreateTodoHandler(Handler):
    """Handler for POST /todos."""

    async def handle(self, request: Request) -> Response:
        # Parse request body
        body = CreateTodoRequest(**request.json())

        # TODO: Implement handler logic
        todo = Todo(
            id=UUID("550e8400-e29b-41d4-a716-446655440000"),
            title=body.title,
            status="pending"
        )

        return Response(
            status_code=201,
            body=todo.__dict__
        )

Integration Example¶

# app.py
from spikard import Spikard
from generated.todo_handlers import ListTodosHandler, CreateTodoHandler

app = Spikard()

# Register generated handlers
app.get("/todos", ListTodosHandler())
app.post("/todos", CreateTodoHandler())

if __name__ == "__main__":
    app.run()

TypeScript Example¶

spikard generate openapi todo-api.openapi.yaml --lang typescript --output ./src/generated/todo.ts

Generated: src/generated/todo.ts

// DO NOT EDIT - Auto-generated by Spikard CLI
'use strict';

export interface Todo {
  id: string;  // UUID format
  title: string;
  status: 'pending' | 'completed';
}

export interface CreateTodoRequest {
  title: string;
}

export interface ListTodosQueryParams {
  status?: 'pending' | 'completed';
}

export async function listTodosHandler(
  request: Request
): Promise<Response> {
  const params = request.query as ListTodosQueryParams;

  // TODO: Implement handler logic
  const todos: Todo[] = [];

  return Response.json({ todos }, { status: 200 });
}

export async function createTodoHandler(
  request: Request
): Promise<Response> {
  const body = await request.json() as CreateTodoRequest;

  // TODO: Implement handler logic
  const todo: Todo = {
    id: crypto.randomUUID(),
    title: body.title,
    status: 'pending',
  };

  return Response.json(todo, { status: 201 });
}

AsyncAPI Code Generation¶

Generate handlers for WebSocket and Server-Sent Events (SSE) from AsyncAPI 3.0 specs.

WebSocket Example¶

spikard generate asyncapi chat-service.asyncapi.yaml --lang python --output ./generated/chat_handlers.py

Schema: chat-service.asyncapi.yaml

asyncapi: 3.0.0
info:
  title: Chat Service
  version: 1.0.0

channels:
  chat:
    address: /chat/{roomId}
    messages:
      chatMessage:
        payload:
          type: object
          required: [text]
          properties:
            text:
              type: string
              maxLength: 500
            replyTo:
              type: string
              format: uuid

      chatMessageBroadcast:
        payload:
          type: object
          properties:
            messageId:
              type: string
              format: uuid
            userId:
              type: string
            text:
              type: string
            timestamp:
              type: string
              format: date-time

Generated Output:

#!/usr/bin/env python3
# DO NOT EDIT - Auto-generated by Spikard CLI

from dataclasses import dataclass
from typing import Optional
from uuid import UUID
from datetime import datetime
from spikard.websocket import WebSocketHandler, WebSocketConnection

@dataclass
class ChatMessage:
    """Incoming chat message from client."""
    text: str
    reply_to: Optional[UUID] = None

@dataclass
class ChatMessageBroadcast:
    """Outgoing chat message to all clients."""
    message_id: UUID
    user_id: str
    text: str
    timestamp: datetime

class ChatHandler(WebSocketHandler):
    """WebSocket handler for /chat/{roomId}."""

    async def on_connect(self, connection: WebSocketConnection) -> None:
        """Called when client connects."""
        room_id = connection.path_params["roomId"]
        # TODO: Add connection to room

    async def on_message(self, connection: WebSocketConnection, data: dict) -> None:
        """Called when client sends a message."""
        message = ChatMessage(**data)

        # TODO: Validate, persist, and broadcast message
        broadcast = ChatMessageBroadcast(
            message_id=UUID("550e8400-e29b-41d4-a716-446655440000"),
            user_id=connection.user_id,
            text=message.text,
            timestamp=datetime.utcnow()
        )

        # Broadcast to all clients in room
        await connection.broadcast(broadcast.__dict__)

    async def on_disconnect(self, connection: WebSocketConnection) -> None:
        """Called when client disconnects."""
        # TODO: Remove connection from room

Server-Sent Events Example¶

spikard generate asyncapi events-stream.asyncapi.yaml --lang typescript --output ./src/handlers/events.ts

Generated Output:

// DO NOT EDIT - Auto-generated by Spikard CLI
'use strict';

export interface SystemAlert {
  severity: 'info' | 'warning' | 'critical';
  message: string;
  timestamp: string;
}

export async function* eventsStreamHandler(
  request: Request
): AsyncGenerator<ServerSentEvent> {
  // Extract query parameters
  const filter = new URL(request.url).searchParams.get('filter');

  // TODO: Subscribe to event source
  while (true) {
    const event: SystemAlert = {
      severity: 'info',
      message: 'System operational',
      timestamp: new Date().toISOString(),
    };

    yield {
      event: 'systemAlert',
      data: JSON.stringify(event),
    };

    await new Promise(resolve => setTimeout(resolve, 5000));
  }
}

GraphQL Code Generation¶

Generate type-safe resolvers and type definitions from GraphQL schemas.

Basic Usage¶

# Generate all targets (types + resolvers)
spikard generate graphql schema.graphql --lang python --target all

# Generate only types (DTOs)
spikard generate graphql schema.graphql --lang typescript --target types

# Generate only resolvers
spikard generate graphql schema.graphql --lang ruby --target resolvers

Example Schema¶

# schema.graphql
type Query {
  user(id: ID!): User
  users(limit: Int = 20, offset: Int = 0): [User!]!
}

type Mutation {
  createUser(input: CreateUserInput!): User!
  updateUser(id: ID!, input: UpdateUserInput!): User
}

type User {
  id: ID!
  name: String!
  email: String!
  role: Role!
  createdAt: DateTime!
}

enum Role {
  ADMIN
  USER
  GUEST
}

input CreateUserInput {
  name: String!
  email: String!
  role: Role = USER
}

input UpdateUserInput {
  name: String
  email: String
  role: Role
}

scalar DateTime

Generated Output (Python)¶

spikard generate graphql schema.graphql --lang python --target types --output ./generated/graphql/types.py

Generated: generated/graphql/types.py

#!/usr/bin/env python3
# DO NOT EDIT - Auto-generated by Spikard CLI

from dataclasses import dataclass
from typing import Optional, List
from datetime import datetime
from enum import Enum

class Role(str, Enum):
    """User role enumeration."""
    ADMIN = "ADMIN"
    USER = "USER"
    GUEST = "GUEST"

@dataclass
class User:
    """A user in the system."""
    id: str
    name: str
    email: str
    role: Role
    created_at: datetime

@dataclass
class CreateUserInput:
    """Input for creating a user."""
    name: str
    email: str
    role: Role = Role.USER

@dataclass
class UpdateUserInput:
    """Input for updating a user."""
    name: Optional[str] = None
    email: Optional[str] = None
    role: Optional[Role] = None

spikard generate graphql schema.graphql --lang python --target resolvers --output ./generated/graphql/resolvers.py

Generated: generated/graphql/resolvers.py

#!/usr/bin/env python3
# DO NOT EDIT - Auto-generated by Spikard CLI

from typing import Optional, List
from .types import User, Role, CreateUserInput, UpdateUserInput

class QueryResolvers:
    """Resolvers for Query type."""

    @staticmethod
    async def user(id: str) -> Optional[User]:
        """Resolve user by ID."""
        # TODO: Implement resolver logic
        return None

    @staticmethod
    async def users(limit: int = 20, offset: int = 0) -> List[User]:
        """Resolve list of users."""
        # TODO: Implement resolver logic
        return []

class MutationResolvers:
    """Resolvers for Mutation type."""

    @staticmethod
    async def create_user(input: CreateUserInput) -> User:
        """Resolve createUser mutation."""
        # TODO: Implement resolver logic
        from datetime import datetime
        return User(
            id="new-user-id",
            name=input.name,
            email=input.email,
            role=input.role,
            created_at=datetime.utcnow()
        )

    @staticmethod
    async def update_user(id: str, input: UpdateUserInput) -> Optional[User]:
        """Resolve updateUser mutation."""
        # TODO: Implement resolver logic
        return None

Integration Example¶

# app.py
from spikard.graphql import GraphQLServer
from generated.graphql.resolvers import QueryResolvers, MutationResolvers

app = GraphQLServer(
    schema_path="schema.graphql",
    query_resolvers=QueryResolvers,
    mutation_resolvers=MutationResolvers
)

if __name__ == "__main__":
    app.run()

OpenRPC Code Generation¶

Generate JSON-RPC 2.0 handlers from OpenRPC specifications.

Basic Usage¶

spikard generate jsonrpc user-api.openrpc.json --lang python --output ./generated/rpc_handlers.py

Example Schema¶

{
  "openrpc": "1.3.2",
  "info": {
    "title": "User Management API",
    "version": "1.0.0"
  },
  "methods": [
    {
      "name": "user.getById",
      "params": [
        {
          "name": "userId",
          "required": true,
          "schema": {
            "type": "string",
            "format": "uuid"
          }
        }
      ],
      "result": {
        "name": "user",
        "schema": {
          "type": "object",
          "properties": {
            "id": { "type": "string" },
            "name": { "type": "string" },
            "email": { "type": "string" }
          }
        }
      }
    }
  ]
}

Generated Output (TypeScript)¶

// DO NOT EDIT - Auto-generated by Spikard CLI
'use strict';

export interface User {
  id: string;
  name: string;
  email: string;
}

export interface JsonRpcRequest {
  jsonrpc: '2.0';
  method: string;
  params?: unknown;
  id?: string | number;
}

export interface JsonRpcResponse {
  jsonrpc: '2.0';
  result?: unknown;
  error?: JsonRpcError;
  id: string | number | null;
}

export interface JsonRpcError {
  code: number;
  message: string;
  data?: unknown;
}

export async function userGetByIdHandler(
  userId: string
): Promise<User> {
  // TODO: Implement RPC method
  return {
    id: userId,
    name: 'Example User',
    email: 'user@example.com',
  };
}

export async function handleJsonRpcRequest(
  request: JsonRpcRequest
): Promise<JsonRpcResponse> {
  try {
    switch (request.method) {
      case 'user.getById': {
        const userId = request.params as string;
        const result = await userGetByIdHandler(userId);
        return { jsonrpc: '2.0', result, id: request.id ?? null };
      }

      default:
        return {
          jsonrpc: '2.0',
          error: {
            code: -32601,
            message: 'Method not found',
          },
          id: request.id ?? null,
        };
    }
  } catch (error) {
    return {
      jsonrpc: '2.0',
      error: {
        code: -32603,
        message: 'Internal error',
      },
      id: request.id ?? null,
    };
  }
}

Protobuf/gRPC Code Generation¶

Generate type-safe gRPC service handlers from .proto files. This is the most detailed section as protobuf requires binary serialization and supports four streaming modes.

Architecture¶

Spikard's protobuf generator produces code that integrates with existing gRPC runtime:

No standalone server: Integrates with Spikard HTTP server
Binary serialization: Uses language-specific protobuf libraries
Four streaming modes: Unary, server streaming, client streaming, bidirectional
Type safety: Strict type mapping with proper nullability

Basic Usage¶

# Generate Python protobuf code
spikard generate protobuf user.proto --lang python --output ./generated/user_pb.py

# Generate TypeScript with a custom output file
spikard generate protobuf api.proto --lang typescript --output ./src/generated/api_pb.ts

# Generate all languages
for lang in python typescript ruby php rust; do
  spikard generate protobuf schema.proto --lang $lang --output ./generated/$lang.pb
done

Example Proto Schema¶

// user_service.proto
syntax = "proto3";

package user.v1;

service UserService {
  // Unary RPC: single request → single response
  rpc GetUser(GetUserRequest) returns (User);

  // Server streaming: single request → stream of responses
  rpc ListUsers(ListUsersRequest) returns (stream User);

  // Client streaming: stream of requests → single response
  rpc CreateUsers(stream CreateUserRequest) returns (CreateUsersResponse);

  // Bidirectional streaming: stream ↔ stream
  rpc ChatWithSupport(stream ChatMessage) returns (stream ChatMessage);
}

message GetUserRequest {
  string user_id = 1;
}

message User {
  string id = 1;
  string name = 2;
  optional string email = 3;  // proto3 optional
  Role role = 4;
  int64 created_at = 5;
}

enum Role {
  ROLE_UNSPECIFIED = 0;
  ROLE_ADMIN = 1;
  ROLE_USER = 2;
  ROLE_GUEST = 3;
}

message ListUsersRequest {
  int32 page = 1;
  int32 page_size = 2;
}

message CreateUserRequest {
  string name = 1;
  string email = 2;
  Role role = 3;
}

message CreateUsersResponse {
  repeated User users = 1;
  int32 created_count = 2;
}

message ChatMessage {
  string user_id = 1;
  string text = 2;
  int64 timestamp = 3;
}

Type Mapping¶

Protobuf types map to language-native types:

Proto3 Type	Python	TypeScript	Ruby	PHP	Rust
`double`	`float`	`number`	`Float`	`float`	`f64`
`float`	`float`	`number`	`Float`	`float`	`f32`
`int32`	`int`	`number`	`Integer`	`int`	`i32`
`int64`	`int`	`bigint`	`Integer`	`int`	`i64`
`bool`	`bool`	`boolean`	`Boolean`	`bool`	`bool`
`string`	`str`	`string`	`String`	`string`	`String`
`bytes`	`bytes`	`Uint8Array`	`String`	`string`	`Bytes`
`repeated T`	`list[T]`	`T[]`	`Array<T>`	`array<T>`	`Vec<T>`
`map<K,V>`	`dict[K,V]`	`Map<K,V>`	`Hash{K=>V}`	`array<K,V>`	`HashMap<K,V>`
`optional T`	`T\\|None`	`T\\|undefined`	`T\\|nil`	`?T`	`Option<T>`

Generated Output (Python)¶

spikard generate protobuf user_service.proto --lang python --output ./generated/user_service.py

Generated: generated/user_service.py

#!/usr/bin/env python3
# DO NOT EDIT - Auto-generated by Spikard CLI
#
# This file was automatically generated from your Protobuf schema.
# Any manual changes will be overwritten on the next generation.
"""Protocol Buffer message definitions."""

from __future__ import annotations

from google.protobuf import message as _message
from google.protobuf import descriptor as _descriptor
from typing import Optional, List
from enum import Enum

# Package: user.v1

class Role(int, Enum):
    """User role enumeration."""
    ROLE_UNSPECIFIED = 0
    ROLE_ADMIN = 1
    ROLE_USER = 2
    ROLE_GUEST = 3

class GetUserRequest(_message.Message):
    """Request to get a user by ID."""
    user_id: str

class User(_message.Message):
    """A user entity."""
    id: str
    name: str
    email: Optional[str]
    role: Role
    created_at: int

class ListUsersRequest(_message.Message):
    """Request to list users with pagination."""
    page: int
    page_size: int

class CreateUserRequest(_message.Message):
    """Request to create a user."""
    name: str
    email: str
    role: Role

class CreateUsersResponse(_message.Message):
    """Response after creating multiple users."""
    users: List[User]
    created_count: int

class ChatMessage(_message.Message):
    """A chat message."""
    user_id: str
    text: str
    timestamp: int

Same generated file: service definitions section

#!/usr/bin/env python3
# DO NOT EDIT - Auto-generated by Spikard CLI
"""Protocol Buffer service definitions."""

from __future__ import annotations

import grpc
from typing import AsyncIterator
from .user_service_pb import (
    GetUserRequest, User, ListUsersRequest,
    CreateUserRequest, CreateUsersResponse, ChatMessage
)

# Package: user.v1

class UserServiceHandler:
    """gRPC handler for UserService."""

    async def GetUser(self, request: GetUserRequest) -> User:
        """Unary RPC: Get a single user by ID."""
        # TODO: Implement handler logic
        return User(
            id=request.user_id,
            name="Example User",
            email="user@example.com",
            role=Role.ROLE_USER,
            created_at=1640000000
        )

    async def ListUsers(
        self, request: ListUsersRequest
    ) -> AsyncIterator[User]:
        """Server streaming RPC: Stream users with pagination."""
        # TODO: Implement handler logic
        for i in range(request.page_size):
            yield User(
                id=f"user-{i}",
                name=f"User {i}",
                email=f"user{i}@example.com",
                role=Role.ROLE_USER,
                created_at=1640000000 + i
            )

    async def CreateUsers(
        self, request_iterator: AsyncIterator[CreateUserRequest]
    ) -> CreateUsersResponse:
        """Client streaming RPC: Create multiple users from stream."""
        created_users: List[User] = []

        # TODO: Implement handler logic
        async for req in request_iterator:
            user = User(
                id=f"new-user-{len(created_users)}",
                name=req.name,
                email=req.email,
                role=req.role,
                created_at=1640000000
            )
            created_users.append(user)

        return CreateUsersResponse(
            users=created_users,
            created_count=len(created_users)
        )

    async def ChatWithSupport(
        self, request_iterator: AsyncIterator[ChatMessage]
    ) -> AsyncIterator[ChatMessage]:
        """Bidirectional streaming RPC: Real-time chat."""
        # TODO: Implement handler logic
        async for msg in request_iterator:
            # Echo back with response
            yield ChatMessage(
                user_id="support-bot",
                text=f"Received: {msg.text}",
                timestamp=msg.timestamp + 1
            )

Generated Output (TypeScript)¶

spikard generate protobuf user_service.proto --lang typescript --output ./src/generated/user_service.ts

Generated: src/generated/user_service.ts

// DO NOT EDIT - Auto-generated by Spikard CLI
'use strict';

import { Message } from 'protobufjs';

// Package: user.v1

export enum Role {
  ROLE_UNSPECIFIED = 0,
  ROLE_ADMIN = 1,
  ROLE_USER = 2,
  ROLE_GUEST = 3,
}

export interface GetUserRequest {
  userId: string;
}

export interface User {
  id: string;
  name: string;
  email?: string;
  role: Role;
  createdAt: bigint;
}

export interface ListUsersRequest {
  page: number;
  pageSize: number;
}

export interface CreateUserRequest {
  name: string;
  email: string;
  role: Role;
}

export interface CreateUsersResponse {
  users: User[];
  createdCount: number;
}

export interface ChatMessage {
  userId: string;
  text: string;
  timestamp: bigint;
}

// Serialization helpers
export function deserializeUser(msg: Message): User {
  return {
    id: msg.id as string,
    name: msg.name as string,
    email: msg.email as string | undefined,
    role: msg.role as Role,
    createdAt: BigInt(msg.createdAt as number),
  };
}

export function serializeUser(user: User): Message {
  // TODO: Implement serialization
  return {} as Message;
}

Generated Output (Rust)¶

spikard generate protobuf user_service.proto --lang rust --output ./src/generated/user_service.rs

Generated: src/generated/user_service.rs

// DO NOT EDIT - Auto-generated by Spikard CLI

use prost::Message;
use bytes::Bytes;
use std::collections::HashMap;

// Package: user.v1

#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
#[repr(i32)]
pub enum Role {
    Unspecified = 0,
    Admin = 1,
    User = 2,
    Guest = 3,
}

#[derive(Clone, PartialEq, Message)]
pub struct GetUserRequest {
    #[prost(string, tag = "1")]
    pub user_id: String,
}

#[derive(Clone, PartialEq, Message)]
pub struct User {
    #[prost(string, tag = "1")]
    pub id: String,

    #[prost(string, tag = "2")]
    pub name: String,

    #[prost(string, optional, tag = "3")]
    pub email: Option<String>,

    #[prost(enumeration = "Role", tag = "4")]
    pub role: i32,

    #[prost(int64, tag = "5")]
    pub created_at: i64,
}

#[derive(Clone, PartialEq, Message)]
pub struct ListUsersRequest {
    #[prost(int32, tag = "1")]
    pub page: i32,

    #[prost(int32, tag = "2")]
    pub page_size: i32,
}

#[derive(Clone, PartialEq, Message)]
pub struct CreateUserRequest {
    #[prost(string, tag = "1")]
    pub name: String,

    #[prost(string, tag = "2")]
    pub email: String,

    #[prost(enumeration = "Role", tag = "3")]
    pub role: i32,
}

#[derive(Clone, PartialEq, Message)]
pub struct CreateUsersResponse {
    #[prost(message, repeated, tag = "1")]
    pub users: Vec<User>,

    #[prost(int32, tag = "2")]
    pub created_count: i32,
}

#[derive(Clone, PartialEq, Message)]
pub struct ChatMessage {
    #[prost(string, tag = "1")]
    pub user_id: String,

    #[prost(string, tag = "2")]
    pub text: String,

    #[prost(int64, tag = "3")]
    pub timestamp: i64,
}

Integration Example (Python)¶

# app.py
from spikard import Spikard
from spikard.grpc import GrpcHandler
from generated.user_service_grpc import UserServiceHandler

app = Spikard()

# Register gRPC service
handler = UserServiceHandler()
app.register_grpc_service("user.v1.UserService", handler)

if __name__ == "__main__":
    app.run()

Testing Generated Code¶

# test_user_service.py
import pytest
from generated.user_service_pb import GetUserRequest, User, Role

@pytest.mark.asyncio
async def test_get_user():
    from generated.user_service_grpc import UserServiceHandler

    handler = UserServiceHandler()
    request = GetUserRequest(user_id="user-123")

    user = await handler.GetUser(request)

    assert user.id == "user-123"
    assert user.name == "Example User"
    assert user.role == Role.ROLE_USER

@pytest.mark.asyncio
async def test_list_users_streaming():
    from generated.user_service_grpc import UserServiceHandler

    handler = UserServiceHandler()
    request = ListUsersRequest(page=1, page_size=5)

    users = []
    async for user in handler.ListUsers(request):
        users.append(user)

    assert len(users) == 5
    assert all(isinstance(u, User) for u in users)

Output Organization¶

Generated files follow language-specific conventions:

Python¶

generated/
├── __init__.py
├── types.py              # DTOs/models
├── handlers.py           # Request handlers
└── graphql/
    ├── __init__.py
    ├── types.py          # GraphQL types
    └── resolvers.py      # GraphQL resolvers

TypeScript¶

src/generated/
├── types.ts              # Interfaces/types
├── handlers.ts           # Handler functions
└── graphql/
    ├── types.ts
    └── resolvers.ts

Ruby¶

lib/generated/
├── types.rb
├── handlers.rb
└── graphql/
    ├── types.rb
    └── resolvers.rb

PHP¶

src/Generated/
├── Types/
│   ├── User.php
│   └── CreateUserRequest.php
├── Handlers/
│   ├── ListUsersHandler.php
│   └── CreateUserHandler.php
└── GraphQL/
    ├── Types/
    └── Resolvers/

Rust¶

src/generated/
├── mod.rs
├── types.rs
├── handlers.rs
└── graphql/
    ├── mod.rs
    ├── types.rs
    └── resolvers.rs

Quality Validation¶

All generated code must pass strict quality tools. Spikard runs these automatically during generation:

Python¶

# Type checking (strictest mode)
mypy --strict generated/

# Linting
ruff check generated/

# Format checking
ruff format --check generated/

Common Issues:

Missing type hints → Add explicit return types
Optional not imported → Generator adds automatically
Mutable default arguments → Use field(default_factory=list)

TypeScript¶

# Type checking
tsc --noEmit

# Linting and formatting
biome check src/generated/

# Alternative: ESLint
eslint src/generated/

Common Issues:

any types → Generator uses strict types
Missing null checks → Uses T | undefined for optionals
Unused imports → Generator only imports what's needed

Ruby¶

# Syntax check
ruby -c lib/generated/handlers.rb

# Type checking
steep check

# Linting
rubocop lib/generated/

Common Issues:

Type signature mismatches → Check RBS files
Missing method definitions → Regenerate code
Style violations → Use rubocop -a to auto-fix

PHP¶

# Syntax check
php -l src/Generated/

# Static analysis (strictest level)
phpstan analyse --level=max src/Generated/

# Code style
php-cs-fixer fix --dry-run src/Generated/

Common Issues:

Missing type declarations → Generator uses strict types
Nullable type handling → Uses ?Type syntax
PSR-12 violations → Generator follows PSR-12

Rust¶

# Compile check
cargo check

# Linting (all warnings as errors)
cargo clippy -- -D warnings

# Format checking
cargo fmt -- --check

Common Issues:

Unused variables → Prefix with _ or remove
Missing trait bounds → Generator adds required bounds
Lifetime issues → Generator uses appropriate lifetimes

Manual Quality Check¶

# Generate to a file
spikard generate openapi schema.yaml --lang python --output ./generated/handlers.py

# Then run project tooling explicitly
python3 -m py_compile ./generated/handlers.py

CLI Options Reference¶

Common Options (All Protocols)¶

spikard generate <protocol> <schema> [OPTIONS]

Options:
  --lang <language>          Target language (python|typescript|ruby|php|rust)
  --output <path>            Output file path
  --help                     Show help message

Protocol-Specific Options¶

OpenAPI¶

spikard generate openapi <schema> [OPTIONS]

Additional Options:
  --dto <style>              DTO style for generated request/response models

GraphQL¶

spikard generate graphql <schema> [OPTIONS]

Additional Options:
  --target <types|resolvers|schema|all>   What to generate (default: all)

Protobuf¶

spikard generate protobuf <proto-file> [OPTIONS]

Additional Options:
  --include <path>           Add import path for dependencies

AsyncAPI¶

spikard generate asyncapi <schema> [OPTIONS]

Examples¶

# Generate only Python DTOs from OpenAPI
spikard generate openapi api.yaml --lang python --dto --output ./models.py

# Generate TypeScript types only from GraphQL
spikard generate graphql schema.graphql --lang typescript --target types --output ./src/graphql/types.ts

# Generate Ruby AsyncAPI handlers
spikard generate asyncapi events.yaml --lang ruby --output ./lib/events.rb

# Generate Rust protobuf code
spikard generate protobuf api.proto --lang rust --output ./src/generated/api.rs

# Generate protobuf code with imported local dependencies
spikard generate protobuf api.proto --lang python --include ./protos --output ./generated/api_pb.py

Troubleshooting¶

Common Issues¶

1. Schema Parsing Errors¶

Problem:

Error: Failed to parse OpenAPI schema: Invalid reference #/components/schemas/User

Solution:

Validate schema with online tools (swagger.io, asyncapi.com)
Check that all $ref paths exist
Ensure schema version matches (OpenAPI 3.1, AsyncAPI 3.0)
Use absolute references within the same file

2. Type Mapping Issues¶

Problem:

Error: Cannot map protobuf type 'google.protobuf.Timestamp' to Python

Solution:

Import well-known types in proto file:

import "google/protobuf/timestamp.proto";

Use built-in proto3 types when possible
Check generator supports the proto type

3. Generated Code Validation Failures¶

Problem:

generated/handlers.py:15: error: Missing return statement

Solution:

Review generated code for TODOs
Implement required handler logic
Run the relevant language toolchain directly against the generated file
Fix type hints if needed

4. Import Errors¶

Problem:

ImportError: cannot import name 'User' from 'generated.types'

Solution:

Check generated files exist at expected paths
Verify __init__.py files exist (Python)
Check module exports (TypeScript/JavaScript)
Regenerate code if structure changed

5. Protobuf Compilation Errors¶

Problem:

Error: user_service.proto:5:10: "User" is not defined

Solution:

Define messages before using them
Check package names match
Add import statements for external protos:

import "common/types.proto";

Use --include flag for import paths:

spikard generate protobuf api.proto --include ./protos --output ./generated/api_pb.py

6. Streaming Type Errors¶

Problem:

Type 'AsyncGenerator<User>' is not assignable to type 'Promise<User[]>'

Solution:

Server streaming returns AsyncGenerator not Promise
Update handler signature:

async function* listUsers(): AsyncGenerator<User>

For client code, collect stream:

const users: User[] = [];
for await (const user of stream) {
  users.push(user);
}

7. Output Path Issues¶

Problem:

Error: Permission denied: ./generated/handlers.py

Solution:

Check write permissions on the parent directory
Pass a file path, not a directory path
Avoid writing to system directories

Debugging Tips¶

Enable verbose output:

RUST_LOG=debug spikard generate openapi api.yaml --lang python

Validate schema separately:

# OpenAPI
npx @apidevtools/swagger-cli validate openapi.yaml

# AsyncAPI
npx @asyncapi/cli validate asyncapi.yaml

Generate to a scratch file for inspection:

spikard generate openapi api.yaml --output test.py

Check generator version:

spikard --version

Use example schemas:

# Test with known-good schema
spikard generate openapi examples/schemas/todo-api.openapi.yaml --lang python

Getting Help¶

Documentation: https://github.com/Goldziher/spikard/docs
Examples: See examples/schemas/ directory
Issues: https://github.com/Goldziher/spikard/issues
ADRs: docs/adr/0004-code-generation.md, docs/adr/0010-protobuf-grpc-code-generation.md

Best Practices¶

Schema Design¶

Use descriptive names: CreateUserRequest not UserInput
Add descriptions: Generates better docstrings
Version your schemas: Include version in filename and package
Keep schemas flat: Avoid deep nesting (max 3 levels)
Use standard formats: uuid, email, date-time, uri

Code Generation¶

Version control generated code: Helps track schema changes
Use CI/CD: Regenerate on schema changes
Don't edit generated files: They'll be overwritten
Keep handlers separate: Implement business logic in separate files
Test generated code: Write integration tests

Project Organization¶

my-api/
├── schemas/              # API schemas (version controlled)
│   ├── openapi.yaml
│   └── user.proto
├── generated/            # Generated code (version controlled)
│   ├── types.py
│   └── handlers.py
├── handlers/             # Custom handler implementations
│   └── user_handlers.py
├── tests/
│   ├── test_generated.py
│   └── test_handlers.py
└── app.py                # Main application

Workflow¶

# 1. Edit schema
vim schemas/api.yaml

# 2. Regenerate code
spikard generate openapi schemas/api.yaml --lang python --output generated/

# 3. Run quality checks
mypy --strict generated/
pytest tests/

# 4. Implement custom logic
vim handlers/user_handlers.py

# 5. Test integration
pytest tests/test_handlers.py

# 6. Commit changes
git add schemas/ generated/ handlers/ tests/
git commit -m "feat: add user endpoints"

Conclusion¶

Spikard's code generator transforms schemas into production-ready, type-safe code across six languages. Generated code:

Passes strict quality tools (mypy --strict, tsc, steep, phpstan level max, clippy)
Integrates with Spikard runtime
Follows language-specific conventions
Provides clear extension points for custom logic

Start with simple schemas, validate early, and use quality tools to catch issues before production.

For more details, see: