# Onyx Development Best Practices: A Comprehensive Guide to Contributing

> Master Onyx development best practices. Learn to use Celery workers, raise OnyxError, enforce typing, and submit concise pull requests with thorough tests for the onyx dot app repository.

- Repository: [Onyx/onyx](https://github.com/onyx-dot-app/onyx)
- Tags: best-practices
- Published: 2026-03-28

---

**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`](https://github.com/onyx-dot-app/onyx/blob/main/contributing_guides/best_practices.md) and [`AGENTS.md`](https://github.com/onyx-dot-app/onyx/blob/main/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:

```python

# 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`](https://github.com/onyx-dot-app/onyx/blob/main/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`](https://github.com/onyx-dot-app/onyx/blob/main/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.

```python

# 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:

```python

# 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`](https://github.com/onyx-dot-app/onyx/blob/main/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.