Onyx Development Best Practices: A Comprehensive Guide to Contributing

When developing with Onyx, use thread-based Celery workers with expiration settings, raise OnyxError instead of HTTPException, enforce strict typing across Python and TypeScript, and keep pull requests under 500 lines with comprehensive test coverage.

Onyx is a modular Gen-AI and Enterprise Search platform built with FastAPI, Celery, PostgreSQL, Redis, and Vespa. The codebase maintains rigorous standards for type safety, error handling, and asynchronous task processing to support its multi-tenant architecture. Following the conventions established in contributing_guides/best_practices.md and AGENTS.md ensures your contributions remain performant, maintainable, and consistent with the project's distributed systems design.

Architectural Patterns for Onyx Development

Worker Design and Task Definition

Onyx relies on thread-based Celery workers organized into specific roles: primary, docfetching, docprocessing, light, heavy, kg_processing, monitoring, user_file_processing, and beat. Each worker must have a clear, bounded responsibility to guarantee stable resource usage and prevent unbounded queues.

Define all tasks in backend/onyx/background/celery/tasks/ using the @shared_task decorator exclusively. Always provide an expires attribute to prevent runaway jobs and ensure tenant isolation:

# backend/onyx/background/celery/tasks/example.py

from celery import shared_task
from datetime import timedelta
from onyx.error_handling.exceptions import OnyxError
from onyx.error_handling.error_codes import OnyxErrorCode

@shared_task(expires=timedelta(minutes=10).total_seconds())
def refresh_connector(connector_id: int) -> None:
    """Refresh a connector – runs in the docfetching worker."""
    try:
        connector = get_connector(connector_id)
        connector.refresh()
    except SomeDomainError as exc:
        raise OnyxError(OnyxErrorCode.BAD_REQUEST, str(exc))

API Design and Error Handling

Do not use response_model in FastAPI endpoints; rely on type hints only to keep the API layer lightweight and avoid unnecessary Pydantic validation overhead. For error handling, raise OnyxError from onyx.error_handling.exceptions instead of HTTPException. Use the provided OnyxErrorCode enum and optionally status_code_override to guarantee a uniform JSON error payload across the service.

Maintain strict separation of concerns in the repository layout:

• Data models → models.py (Pydantic)
• DB helpers → backend/onyx/db/
• Prompts → prompts/
• API routers → backend/onyx/server/

Core Engineering Principles

Strict Typing and Type Safety

Use strict typing everywhere in both Python and TypeScript. Prefer cast only when Any appears, and favor domain-specific Pydantic models over opaque structures like dictionaries. This approach improves readability and catches errors at static analysis time rather than runtime.

Design APIs that are hard to misuse by eliminating duplicated logic early and validating objects on creation. Keep configuration objects free of DB sessions, favor composition over inheritance, and avoid mutable global state.

Error Handling Philosophy

Fail loudly—let exceptions propagate rather than swallowing errors with broad try/except blocks. When modifying code that violates a best practice, fix that issue in the same PR (but avoid refactoring the entire repository). Add comments at logical boundaries, for assumptions, and for complex regexes to maintain context for future contributors.

Async Caution

Do not add new async code unless there is a proven benefit; keep existing async code synchronous when possible. This constraint prevents complexity in the already-concurrent Celery worker environment and avoids subtle race conditions in the multi-threaded architecture.

Database and Performance Guidelines

SQLAlchemy Best Practices

Prefer eager loading (selectinload) over lazy loading to avoid hidden database queries and session-scope bugs. Do not hold DB sessions or locks longer than needed, and avoid unbounded in-memory structures in connectors. According to the performance guidelines in contributing_guides/best_practices.md, validate objects on creation and before use to maintain correctness by construction.

Module-Level Side Effects

Avoid executing logic at import time. Keep such code behind if __name__ == "__main__": guards or in dedicated scripts to prevent unintended initialization when modules are imported for testing.

Testing Strategy

Onyx employs a four-tier testing strategy to guarantee coverage at every level:

1. Unit tests – Pure Python with no external services, using mocks for dependencies like LLM providers.
2. External-dependency unit tests – Require running services (Postgres, Redis, Vespa, OpenAI).
3. Integration tests – Run against a full stack deployment.
4. Playwright E2E – Frontend and backend validation.

# backend/tests/unit/onyx/llm/test_embedding.py

from unittest.mock import patch
import pytest
from onyx.llm.embeddings import embed_text

@patch("onyx.llm.providers.openai.embed")
def test_embed_text(mock_embed):
    mock_embed.return_value = [0.1, 0.2, 0.3]
    result = embed_text("hello")
    assert result == [0.1, 0.2, 0.3]
    mock_embed.assert_called_once_with("hello")

Code Quality and CI/CD

Run pre-commit install && pre-commit run --all-files before pushing to enforce linting and strict typing checks. Follow trunk-based development by keeping PRs ≤ 500 lines, merging frequently, and using short-lived feature flags. When using feature flags, test both states to ensure safe rollouts:

# backend/onyx/server/api/chat.py

from fastapi import APIRouter
from onyx.utils.feature_flags import is_enabled

router = APIRouter()

@router.get("/chat")
def get_chat(user_id: str):
    if is_enabled("new_chat_flow"):
        return new_chat_handler(user_id)
    else:
        return old_chat_handler(user_id)

Differentiate between one-way doors (irreversible changes requiring deliberation) and two-way doors (reversible changes where you can move fast). Name variables explicitly and descriptively, avoiding single-character identifiers unless trivial.

Summary

• Use @shared_task with expires= for all Celery tasks in backend/onyx/background/celery/tasks/
• Raise OnyxError from onyx.error_handling.exceptions instead of HTTPException for uniform error payloads
• Enforce strict typing and prefer domain-specific Pydantic models over generic structures
• Prefer selectinload over lazy loading in SQLAlchemy to avoid N+1 queries
• Keep PRs under 500 lines and use short-lived feature flags with trunk-based development
• Never use response_model in FastAPI endpoints; rely on type hints only
• Fail loudly by letting exceptions propagate rather than catching broadly

Frequently Asked Questions

Should I use async/await when developing with Onyx?

Avoid adding new async code unless there is a proven performance benefit. The codebase prefers synchronous execution within Celery workers to prevent complexity and subtle concurrency bugs. Keep existing async implementations synchronous when possible, as documented in the async guidelines of contributing_guides/best_practices.md.

How should I handle errors in Onyx API endpoints?

Import OnyxError from onyx.error_handling.exceptions and raise it with an appropriate OnyxErrorCode enum value. This ensures the error handling middleware in backend/onyx/error_handling/ returns a consistent JSON payload. Use status_code_override only when necessary to deviate from default HTTP status mappings.

What is the recommended size for pull requests in Onyx?

Keep pull requests under 500 lines of code to facilitate rapid review and reduce integration risk. Follow trunk-based development practices by merging frequently into the main branch. If introducing feature flags, ensure both enabled and disabled states are tested before the flag is removed.

How do I add a new Celery worker task to Onyx?

Place the task file under backend/onyx/background/celery/tasks/ and decorate functions with @shared_task, not Celery app-specific decorators. Always include an expires parameter to prevent task accumulation, and assign the task to the appropriate worker queue (primary, docfetching, docprocessing, etc.) based on resource requirements.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}