Python Skills

Python Skills for LlamaFarm

Shared Python best practices and code review checklists for all Python components in the LlamaFarm monorepo.

Applicable Components

Quick Reference

Code Style

LlamaFarm uses ruff with shared configuration in ruff.toml:

line-length = 88
target-version = "py311"
select = ["E", "F", "I", "B", "UP", "SIM"]

Key rules:

• E, F: Core pyflakes and pycodestyle
• I: Import sorting (isort)
• B: Bugbear (common pitfalls)
• UP: Upgrade syntax to modern Python
• SIM: Simplify code patterns

Architecture Patterns

Settings with pydantic-settings

from pydantic_settings import BaseSettings

class Settings(BaseSettings, env_file=".env"):
    LOG_LEVEL: str = "INFO"
    HOST: str = "0.0.0.0"
    PORT: int = 14345

settings = Settings()  # Singleton at module level

Structured Logging with structlog

from core.logging import FastAPIStructLogger  # Server
from core.logging import RAGStructLogger      # RAG
from core.logging import UniversalRuntimeLogger  # Runtime

logger = FastAPIStructLogger(__name__)
logger.info("Operation completed", extra={"count": 10, "duration_ms": 150})

Abstract Base Classes for Extensibility

from abc import ABC, abstractmethod

class Component(ABC):
    def __init__(self, name: str, config: dict[str, Any] | None = None):
        self.name = name or self.__class__.__name__
        self.config = config or {}

    @abstractmethod
    def process(self, documents: list[Document]) -> ProcessingResult:
        pass

Dataclasses for Internal Data

from dataclasses import dataclass, field

@dataclass
class Document:
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)
    id: str = field(default_factory=lambda: str(uuid.uuid4()))

Pydantic Models for API Boundaries

from pydantic import BaseModel, Field, ConfigDict

class EmbeddingRequest(BaseModel):
    model: str
    input: str | list[str]
    encoding_format: Literal["float", "base64"] | None = "float"

    model_config = ConfigDict(str_strip_whitespace=True)

Directory Structure

Each Python component follows this structure:

component/
├── pyproject.toml     # UV-managed dependencies
├── core/              # Core functionality
│   ├── __init__.py
│   ├── settings.py    # Pydantic Settings
│   └── logging.py     # structlog setup
├── services/          # Business logic (server)
├── models/            # ML models (runtime)
├── tasks/             # Celery tasks (rag)
├── utils/             # Utility functions
└── tests/
    ├── conftest.py    # Shared fixtures
    └── test_*.py

Review Checklist Summary

When reviewing Python code in LlamaFarm:

1. Patterns (Medium priority)

• Modern Python syntax (3.10+ type hints)
• Dataclass vs Pydantic used appropriately
• No mutable default arguments

2. Async (High priority)

• No blocking calls in async functions
• Proper asyncio.Lock usage
• Cancellation handled correctly

3. Typing (Medium priority)

• Complete return type hints
• Generic types parameterized
• Pydantic v2 patterns

4. Testing (Medium priority)

• Fixtures properly scoped
• Async tests use pytest-asyncio
• Mocks cleaned up

5. Errors (High priority)

• Custom exceptions with context
• Structured logging with extra dict
• Proper exception chaining

6. Security (Critical priority)

• Path traversal prevention
• Input sanitization
• Safe deserialization

See individual topic files for detailed checklists with grep patterns.

Async/Await Best Practices

Async patterns for LlamaFarm's FastAPI and Celery-based services.

---

Category: Async Function Basics

No Blocking Calls in Async Functions

What to check: Async functions must not call blocking I/O

Bad pattern:

async def load_model(model_id: str):
    with open(file_path, "rb") as f:  # BLOCKING
        data = f.read()
    time.sleep(1)  # BLOCKING
    requests.get(url)  # BLOCKING

Good pattern:

async def load_model(model_id: str):
    async with aiofiles.open(file_path, "rb") as f:
        data = await f.read()
    await asyncio.sleep(1)
    async with httpx.AsyncClient() as client:
        response = await client.get(url)

Search pattern:

rg "async def" -A 20 --type py | rg "time\.sleep|requests\.(get|post)|open\("

Pass criteria: No blocking calls inside async functions

Severity: Critical

Recommendation: Use asyncio.to_thread() if blocking is unavoidable

---

Use asyncio.to_thread for CPU-Bound Work

What to check: Offload CPU-bound operations to thread pool

Good pattern (from runtimes/universal):

async def load(self) -> None:
    # Model loading is CPU-bound, run in thread pool
    self.model = await asyncio.to_thread(
        AutoModelForCausalLM.from_pretrained,
        self.model_id,
        trust_remote_code=True
    )

Search pattern:

rg "asyncio\.to_thread|run_in_executor" --type py

Severity: High

---

Async Context Managers

What to check: Use async context managers for async resources

Good pattern:

async with httpx.AsyncClient() as client:
    response = await client.get(url)

async with aiofiles.open(path) as f:
    content = await f.read()

Search pattern:

rg "async with" --type py

Severity: Medium

---

Category: Concurrency Control

asyncio.Lock for Shared State

What to check: Use locks to prevent race conditions

Good pattern (from runtimes/universal/server.py):

_model_load_lock = asyncio.Lock()

async def load_language(model_id: str):
    if cache_key not in _models:
        async with _model_load_lock:
            # Double-check after acquiring lock
            if cache_key not in _models:
                model = LanguageModel(model_id, device)
                await model.load()
                _models[cache_key] = model
    return _models.get(cache_key)

Pass criteria: Double-checked locking pattern for lazy initialization

Severity: High

---

Avoid Global Lock Contention

What to check: Use fine-grained locks or lock-free structures

Bad pattern:

global_lock = asyncio.Lock()

async def any_operation():
    async with global_lock:  # Bottleneck for all operations
        ...

Good pattern:

# Per-resource locks
_resource_locks: dict[str, asyncio.Lock] = {}

async def get_resource_lock(resource_id: str) -> asyncio.Lock:
    if resource_id not in _resource_locks:
        _resource_locks[resource_id] = asyncio.Lock()
    return _resource_locks[resource_id]

Severity: Medium

---

Category: Task Management

Background Tasks with asyncio.create_task

What to check: Long-running background work should use create_task

Good pattern (from runtimes/universal/server.py):

@asynccontextmanager
async def lifespan(app: FastAPI):
    global _cleanup_task
    _cleanup_task = asyncio.create_task(_cleanup_idle_models())
    yield
    if _cleanup_task is not None:
        _cleanup_task.cancel()
        with suppress(asyncio.CancelledError):
            await _cleanup_task

Severity: Medium

---

Handle Task Cancellation

What to check: Background tasks must handle CancelledError

Good pattern:

async def _cleanup_idle_models() -> None:
    while True:
        try:
            await asyncio.sleep(CLEANUP_CHECK_INTERVAL)
            # ... cleanup logic
        except asyncio.CancelledError:
            logger.info("Cleanup task cancelled")
            break  # Exit cleanly
        except Exception as e:
            logger.error(f"Error in cleanup: {e}")
            # Continue running despite errors

Search pattern:

rg "asyncio\.CancelledError|CancelledError" --type py

Pass criteria: CancelledError caught and handled gracefully

Severity: High

---

Use suppress for Expected Cancellation

What to check: Use contextlib.suppress for expected cancellation

Good pattern:

from contextlib import suppress

# When stopping a task
task.cancel()
with suppress(asyncio.CancelledError):
    await task

Severity: Low

---

Category: Async Generators

AsyncGenerator Return Type

What to check: Async generators should have proper type hints

Good pattern (from server/agents/base/agent.py):

from collections.abc import AsyncGenerator

async def run_async_stream(
    self,
    messages: list[LFChatCompletionMessageParam] | None = None,
) -> AsyncGenerator[LFChatCompletionChunk]:
    async for chunk in self._client.stream_chat(messages=messages):
        yield chunk

Search pattern:

rg "AsyncGenerator\[" --type py

Severity: Medium

---

Async For Loops

What to check: Use async for with async iterables

Good pattern:

async for chunk in self._client.stream_chat(messages=messages):
    yield chunk

Bad pattern:

# Collecting all items defeats streaming
chunks = [chunk async for chunk in stream]  # Avoid if streaming is intended

Severity: Medium

---

Category: Concurrent Execution

asyncio.gather for Parallel Operations

What to check: Use gather for independent async operations

Good pattern:

results = await asyncio.gather(
    fetch_data(url1),
    fetch_data(url2),
    fetch_data(url3),
    return_exceptions=True  # Don't fail fast on single error
)

# Handle results
for result in results:
    if isinstance(result, Exception):
        logger.error(f"Request failed: {result}")
    else:
        process(result)

Search pattern:

rg "asyncio\.gather" --type py

Severity: Medium

---

TaskGroup for Structured Concurrency (Python 3.11+)

What to check: Consider TaskGroup for better error handling

Good pattern:

async def process_batch(items: list[str]):
    async with asyncio.TaskGroup() as tg:
        for item in items:
            tg.create_task(process_item(item))
    # All tasks completed successfully if we reach here

Note: TaskGroup raises ExceptionGroup on failure

Severity: Low

---

Category: FastAPI Specific

Async Route Handlers

What to check: Use async def for I/O-bound route handlers

Good pattern:

@app.post("/v1/embeddings")
async def create_embeddings(request: EmbeddingRequest):
    model = await load_encoder(request.model)
    embeddings = await model.embed(texts)
    return {"data": embeddings}

Severity: Medium

---

Lifespan Context Manager

What to check: Use lifespan for startup/shutdown logic

Good pattern (from runtimes/universal/server.py):

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    logger.info("Starting server")
    _cleanup_task = asyncio.create_task(_cleanup_idle_models())

    yield  # Server is running

    # Shutdown
    logger.info("Shutting down")
    _cleanup_task.cancel()
    with suppress(asyncio.CancelledError):
        await _cleanup_task

app = FastAPI(lifespan=lifespan)

Search pattern:

rg "@asynccontextmanager" --type py -A 10 | rg "lifespan"

Severity: Medium

---

Dependency Injection with Async

What to check: FastAPI dependencies can be async

Good pattern:

async def get_model(model_id: str = Query(...)) -> BaseModel:
    return await load_model(model_id)

@app.post("/predict")
async def predict(model: BaseModel = Depends(get_model)):
    return await model.predict()

Severity: Low

---

Category: Celery Integration

Async in Celery Tasks

What to check: Celery tasks are synchronous by default

Pattern (from rag/tasks):

@app.task(bind=True, base=IngestTask)
def ingest_file_with_rag_task(self, project_dir: str, ...):
    # Celery tasks run in worker process, not async
    # Use synchronous code here
    handler = IngestHandler(config_path=str(config_path))
    result = handler.ingest_file(file_data=file_data)
    return result

Note: For async operations in Celery, use asyncio.run() inside the task

Severity: Medium

---

Calling Celery from Async Code

What to check: Use .delay() or .apply_async() (they don't block)

Good pattern:

async def start_ingestion(files: list[str]):
    # apply_async is non-blocking
    result = ingest_file_task.apply_async(args=[file_path])
    return result.id

Severity: Medium

---

Category: Error Handling

Async Exception Handling

What to check: Handle exceptions in async code appropriately

Good pattern:

async def safe_operation():
    try:
        result = await risky_operation()
    except SomeAsyncError as e:
        logger.error("Operation failed", extra={"error": str(e)})
        raise HTTPException(status_code=500, detail=str(e)) from e

Severity: High

---

Timeout for Async Operations

What to check: Use timeouts for external calls

Good pattern:

try:
    result = await asyncio.wait_for(
        external_api_call(),
        timeout=30.0
    )
except asyncio.TimeoutError:
    logger.error("External API timed out")
    raise HTTPException(status_code=504, detail="Gateway timeout")

Search pattern:

rg "asyncio\.wait_for|asyncio\.timeout" --type py

Severity: High

Error Handling Checklist

Exception handling and logging patterns for LlamaFarm Python components.

---

Category: Custom Exceptions

Domain-Specific Exception Classes

What to check: Define custom exceptions for domain errors

Good pattern (from config/helpers/loader.py):

class ConfigError(Exception):
    """Raised when there's an error loading or validating configuration."""
    pass

class ValidationError(ConfigError):
    """Raised when configuration validation fails."""
    def __init__(self, message: str, path: str = ""):
        self.path = path
        super().__init__(f"{message}" + (f" at path {path}" if path else ""))

Good pattern (from rag/utils/embedding_safety.py):

class EmbedderUnavailableError(Exception):
    """Raised when an embedder cannot complete requests."""
    pass

class CircuitBreakerOpenError(Exception):
    """Raised when circuit breaker prevents requests."""
    def __init__(self, message: str, failures: int = 0, reset_time: float = 0):
        self.failures = failures
        self.reset_time = reset_time
        super().__init__(message)

Search pattern:

rg "class \w+Error\(|class \w+Exception\(" --type py

Pass criteria: Domain errors have specific exception classes with context

Severity: Medium

---

Exception Hierarchy

What to check: Exceptions follow a clear hierarchy

Good pattern:

class LlamaFarmError(Exception):
    """Base exception for all LlamaFarm errors."""
    pass

class ConfigError(LlamaFarmError):
    """Configuration-related errors."""
    pass

class ProcessingError(LlamaFarmError):
    """Processing pipeline errors."""
    pass

class EmbeddingError(ProcessingError):
    """Embedding generation errors."""
    pass

Pass criteria: Custom exceptions inherit from a base application exception

Severity: Low

---

Category: Exception Handling

Catch Specific Exceptions

What to check: Catch specific exceptions, not bare Exception

Bad pattern:

try:
    process()
except Exception:  # Too broad
    pass

try:
    process()
except:  # Even worse - catches SystemExit, KeyboardInterrupt
    pass

Good pattern (from server/services/dataset_service.py):

try:
    with open(file_path) as f:
        metadata = MetadataFileContent.model_validate_json(f.read())
        files.append(metadata)
except OSError as e:
    logger.warning(
        "Failed to read metadata file",
        namespace=namespace,
        file=file,
        error=str(e),
    )
except ValueError as e:
    # Pydantic validation errors
    logger.warning(
        "Failed to parse metadata file",
        file=file,
        error=str(e),
    )

Search pattern:

rg "except Exception:|except:" --type py

Pass criteria: Specific exceptions caught; broad catches only at top level

Severity: High

---

Exception Chaining with from

What to check: Chain exceptions to preserve context

Good pattern (from config/helpers/loader.py):

try:
    jsonschema.validate(config, schema)
except jsonschema.ValidationError as e:
    path_str = ".".join(str(p) for p in e.path)
    raise ConfigError(
        f"Configuration validation error: {e.message}"
        + (f" at path {path_str}" if path_str else "")
    ) from e
except Exception as e:
    raise ConfigError(f"Error during validation: {e}") from e

Search pattern:

rg "raise \w+ from " --type py

Pass criteria: Use from e when re-raising to preserve traceback

Severity: Medium

---

No Silent Failures

What to check: Exceptions must not be silently swallowed

Bad pattern:

try:
    risky_operation()
except Exception:
    pass  # Silent failure - data loss, debugging nightmare

Good pattern:

try:
    risky_operation()
except SpecificError as e:
    logger.warning(
        "Operation failed, using fallback",
        extra={"error": str(e), "fallback": fallback_value}
    )
    return fallback_value

Search pattern:

rg "except.*:\s*$" -A 1 --type py | rg "pass$"

Pass criteria: No except: pass without logging or explicit justification

Severity: High

---

Use contextlib.suppress for Expected Exceptions

What to check: Use suppress() when exception is truly expected and ignorable

Good pattern (from server/services/dataset_service.py):

from contextlib import suppress

with suppress(FileNotFoundError):
    existing_file = DataService.get_data_file_metadata_by_hash(
        namespace, project, dataset, file_hash
    )

When to use:

• File may not exist and that's OK
• Optional cleanup that shouldn't fail the operation
• Checking for presence of optional features

Severity: Low

---

Category: Structured Logging

Log with Extra Dict

What to check: Use structured logging with extra dict, not f-strings

Good pattern (from server/services/dataset_service.py):

logger.info(
    "Deleted chunks from vector store",
    namespace=namespace,
    project=project,
    dataset=dataset,
    file_hash=file_hash[:16] + "...",
    deleted_chunks=result.get("deleted_count", 0),
)

logger.warning(
    "Dataset metadata directory not found",
    namespace=namespace,
    project=project,
    dataset=dataset,
    path=dataset_meta_dir,
)

Bad pattern:

logger.info(f"Deleted {count} chunks from {dataset}")  # Hard to parse/search

Search pattern:

rg 'logger\.(info|error|warning|debug)\(f"' --type py

Pass criteria: Structured data in extra dict, not embedded in message

Severity: Medium

---

Log Before Raising

What to check: Log errors with context before raising

Good pattern:

try:
    result = process(data)
except ProcessingError as e:
    logger.error(
        "Processing failed",
        extra={
            "error": str(e),
            "error_type": type(e).__name__,
            "data_id": data.id,
            "operation": "process",
        },
        exc_info=True,
    )
    raise

Severity: Medium

---

Use exc_info for Stack Traces

What to check: Include stack traces for unexpected errors

Good pattern:

try:
    complex_operation()
except Exception as e:
    logger.error(
        "Unexpected error in complex operation",
        extra={"error": str(e)},
        exc_info=True,  # Includes full traceback
    )
    raise

Severity: Medium

---

Error Context in Logs

What to check: Include relevant context for debugging

Good pattern:

logger.error(
    "Document processing failed",
    extra={
        "document_id": doc.id,
        "document_source": doc.source,
        "error_type": type(e).__name__,
        "error_message": str(e),
        "stage": "parsing",
        "retry_count": retry_count,
    }
)

Essential context:

• Entity IDs (document_id, user_id, task_id)
• Operation being performed
• Error type and message
• Current state (stage, retry count)

Severity: Medium

---

Category: FastAPI Error Handling

HTTPException for API Errors

What to check: Use HTTPException for HTTP error responses

Good pattern:

from fastapi import HTTPException

@app.get("/items/{item_id}")
async def get_item(item_id: str):
    item = await fetch_item(item_id)
    if item is None:
        raise HTTPException(
            status_code=404,
            detail=f"Item {item_id} not found"
        )
    return item

@app.post("/process")
async def process_data(request: ProcessRequest):
    try:
        result = await process(request.data)
    except ValidationError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except ProcessingError as e:
        raise HTTPException(status_code=500, detail="Processing failed")
    return result

Severity: High

---

Custom Exception Handlers

What to check: Register handlers for custom exceptions

Good pattern:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(ConfigError)
async def config_error_handler(request: Request, exc: ConfigError):
    return JSONResponse(
        status_code=400,
        content={
            "detail": str(exc),
            "error_code": "CONFIG_ERROR",
            "path": getattr(exc, "path", None),
        }
    )

@app.exception_handler(EmbedderUnavailableError)
async def embedder_error_handler(request: Request, exc: EmbedderUnavailableError):
    return JSONResponse(
        status_code=503,
        content={
            "detail": "Embedding service temporarily unavailable",
            "error_code": "EMBEDDER_UNAVAILABLE",
            "retry_after": 60,
        }
    )

Severity: Medium

---

Error Response Models

What to check: Define consistent error response schemas

Good pattern:

from pydantic import BaseModel

class ErrorResponse(BaseModel):
    detail: str
    error_code: str | None = None
    path: str | None = None

class ValidationErrorResponse(BaseModel):
    detail: list[dict[str, Any]]
    error_code: str = "VALIDATION_ERROR"

Severity: Low

---

Category: Resource Cleanup

Finally for Cleanup

What to check: Use finally for guaranteed cleanup

Good pattern:

resource = acquire_resource()
try:
    result = process(resource)
    return result
finally:
    resource.release()

Severity: High

---

Context Managers for Resources

What to check: Prefer context managers over try/finally

Good pattern (from config/helpers/loader.py):

with open(file_path, encoding="utf-8") as f:
    return yaml_instance.load(f)

async with httpx.AsyncClient() as client:
    response = await client.get(url)

Pass criteria: Use context managers for file handles, connections, locks

Severity: Medium

---

Async Context Managers

What to check: Use async context managers for async cleanup

Good pattern:

from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_model(model_id: str):
    model = await load_model(model_id)
    try:
        yield model
    finally:
        await model.unload()

# Usage
async with managed_model("gpt-2") as model:
    result = await model.generate(prompt)

Severity: Medium

---

Category: Resilience Patterns

Circuit Breaker

What to check: Use circuit breaker for external dependencies

Good pattern (from rag/core/base.py):

class Embedder(Component):
    DEFAULT_FAILURE_THRESHOLD = 5
    DEFAULT_RESET_TIMEOUT = 60.0

    def __init__(self, ...):
        circuit_config = (config or {}).get("circuit_breaker", {})
        self._circuit_breaker = CircuitBreaker(
            failure_threshold=circuit_config.get(
                "failure_threshold", self.DEFAULT_FAILURE_THRESHOLD
            ),
            reset_timeout=circuit_config.get(
                "reset_timeout", self.DEFAULT_RESET_TIMEOUT
            ),
        )

    def check_circuit_breaker(self) -> None:
        if not self._circuit_breaker.can_execute():
            state_info = self._circuit_breaker.get_state_info()
            raise CircuitBreakerOpenError(
                f"Circuit breaker is open for {self.name}. "
                f"Too many consecutive failures."
            )

Severity: Medium (for external service calls)

---

Retry with Backoff

What to check: Use exponential backoff for transient failures

Good pattern:

import backoff

@backoff.on_exception(
    backoff.expo,
    (ConnectionError, TimeoutError),
    max_tries=3,
    max_time=30,
)
async def call_external_api():
    async with httpx.AsyncClient() as client:
        return await client.get(url)

Or manual implementation:

async def with_retry(func, max_retries=3, base_delay=1.0):
    for attempt in range(max_retries):
        try:
            return await func()
        except TransientError as e:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            logger.warning(f"Retry {attempt + 1}/{max_retries} after {delay}s")
            await asyncio.sleep(delay)

Severity: Medium

---

Graceful Degradation

What to check: Provide fallbacks when services fail

Good pattern:

async def get_embeddings(texts: list[str]) -> list[list[float]]:
    try:
        return await primary_embedder.embed(texts)
    except EmbedderUnavailableError:
        logger.warning("Primary embedder unavailable, using fallback")
        try:
            return await fallback_embedder.embed(texts)
        except EmbedderUnavailableError:
            logger.error("All embedders unavailable")
            raise

Severity: Medium

---

Category: Validation Errors

Early Validation

What to check: Validate input at API boundaries

Good pattern:

def process_file(path: str) -> ProcessingResult:
    # Validate early
    file_path = Path(path)
    if not file_path.exists():
        raise FileNotFoundError(f"File not found: {path}")
    if not file_path.is_file():
        raise ValueError(f"Not a file: {path}")
    if file_path.stat().st_size == 0:
        raise ValueError(f"Empty file: {path}")

    # Process validated input
    return _do_process(file_path)

Severity: Medium

---

Pydantic Validation

What to check: Use Pydantic for input validation

Good pattern:

from pydantic import BaseModel, Field, field_validator

class ProcessRequest(BaseModel):
    content: str = Field(min_length=1, max_length=100000)
    options: dict[str, Any] = Field(default_factory=dict)

    @field_validator("content")
    @classmethod
    def validate_content(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Content cannot be blank")
        return v

Severity: High

Python Patterns Checklist

Idiomatic Python patterns for LlamaFarm components.

---

Category: Code Organization

Import Organization

What to check: Imports should be organized in groups: stdlib, third-party, local

Search pattern:

rg "^import |^from " --type py server/ | head -50

Pass criteria: Imports sorted by ruff (isort rules enabled)

Severity: Low

Recommendation: Run ruff check --fix to auto-sort imports

---

Module Structure

What to check: Modules should have clear separation of concerns

Pass criteria:

• Models/types in dedicated models/ or types.py
• Routes in dedicated routers/ or api/
• Business logic in services/ or core/
• Utilities in utils/
• Configuration in core/settings.py

Severity: Medium

---

Category: Data Structures

Dataclass vs Pydantic Selection

What to check: Use the right tool for the job

Pass criteria:

• @dataclass for simple internal data containers (e.g., Document, ProcessingResult)
• pydantic.BaseModel for API request/response models requiring validation
• pydantic_settings.BaseSettings for environment-based configuration

Search pattern:

rg "@dataclass|class.*\(BaseModel\)|class.*\(BaseSettings\)" --type py

Severity: Medium

Example (from rag/core/base.py):

@dataclass
class Document:
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)
    id: str = field(default_factory=lambda: str(uuid.uuid4()))

---

Default Factory for Mutable Defaults

What to check: Never use mutable default arguments

Bad pattern:

def process(items: list = []):  # WRONG - shared mutable default
def init(config: dict = {}):    # WRONG

Good pattern:

def process(items: list | None = None):
    items = items if items is not None else []

# Or with dataclass
@dataclass
class Config:
    items: list[str] = field(default_factory=list)

Search pattern:

rg "def \w+\([^)]*=\s*\[\]|=\s*\{\}" --type py

Pass criteria: No mutable default arguments

Severity: High

---

Field Default Factory in Dataclasses

What to check: Use field(default_factory=...) for mutable defaults

Good pattern:

from dataclasses import dataclass, field

@dataclass
class ProcessingResult:
    documents: list[Document]
    errors: list[dict[str, Any]] = field(default_factory=list)
    metrics: dict[str, Any] = field(default_factory=dict)

Severity: High

---

Category: Modern Python Syntax

Use Modern Type Syntax (PEP 604, 585)

What to check: Use Python 3.10+ type syntax

Pass criteria:

• list[str] not List[str]
• dict[str, Any] not Dict[str, Any]
• str | None not Optional[str]
• tuple[int, ...] not Tuple[int, ...]

Search pattern:

rg "from typing import.*(List|Dict|Tuple|Optional)" --type py

Severity: Low (handled by ruff UP rules)

Recommendation: Run ruff check --fix --select UP to auto-upgrade

---

Use Context Managers for Resources

What to check: Resources should use context managers

Pass criteria:

• File operations use with open()
• Lock acquisition uses async with lock:
• Temporary resources use context managers

Good pattern (from config/helpers/loader.py):

with open(file_path, encoding="utf-8") as f:
    return yaml_instance.load(f)

Search pattern (find potential issues):

rg "\.close\(\)|\.release\(\)" --type py

Severity: Medium

---

Comprehensions over Loops

What to check: Prefer comprehensions for simple transformations

Good pattern:

# List comprehension
texts = [doc.content for doc in documents]

# Dict comprehension
result = {k: _commented_map_to_dict(v) for k, v in obj.items()}

# Generator for large sequences
data = (process(item) for item in large_list)

Bad pattern:

texts = []
for doc in documents:
    texts.append(doc.content)

Severity: Low

---

Use suppress() for Expected Exceptions

What to check: Use contextlib.suppress for expected exceptions

Good pattern (from server/services/dataset_service.py):

from contextlib import suppress

with suppress(FileNotFoundError):
    existing_file = DataService.get_data_file_metadata_by_hash(...)

Bad pattern:

try:
    existing_file = DataService.get_data_file_metadata_by_hash(...)
except FileNotFoundError:
    pass

Severity: Low

---

Category: Pydantic Patterns

Pydantic v2 Validator Syntax

What to check: Use Pydantic v2 validator syntax

Good pattern:

from pydantic import field_validator, model_validator

class Model(BaseModel):
    name: str

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str) -> str:
        return v.strip()

    @model_validator(mode="after")
    def validate_model(self) -> "Model":
        return self

Bad pattern (Pydantic v1):

@validator("name")  # DEPRECATED
@root_validator     # DEPRECATED

Search pattern:

rg "@validator|@root_validator" --type py

Pass criteria: Use @field_validator and @model_validator

Severity: Medium

Known Technical Debt: The RAG module (rag/components/metadata/metadata_config.py) still contains Pydantic v1 @validator decorators and .dict() calls that need migration to v2 patterns. This is tracked for future cleanup.

---

Pydantic v2 Model Config

What to check: Use ConfigDict instead of inner Config class

Good pattern:

from pydantic import BaseModel, ConfigDict

class Model(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True,
        arbitrary_types_allowed=True
    )

Bad pattern (Pydantic v1):

class Model(BaseModel):
    class Config:  # DEPRECATED
        arbitrary_types_allowed = True

Severity: Low

---

Use model_dump() Instead of dict()

What to check: Use Pydantic v2 serialization methods

Good pattern:

config_dict = config.model_dump(mode="json", exclude_none=True)
json_str = model.model_dump_json()

Bad pattern:

config_dict = config.dict()  # DEPRECATED

Search pattern:

rg "\.dict\(\)" --type py | rg -v "model_dump"

Severity: Medium

---

Category: String Formatting

F-Strings for Interpolation

What to check: Use f-strings over .format() or %

Search pattern:

rg "\.format\(|%s|%d" --type py | rg -v "strftime"

Pass criteria: Use f-strings for string interpolation

Severity: Low

---

Structured Logging with Extra Dict

What to check: Use structured logging with extra dict, not f-strings

Good pattern:

logger.info(
    "Processing file",
    extra={"path": path, "size": size, "task_id": task_id}
)

Bad pattern:

logger.info(f"Processing file {path} with size {size}")

Search pattern:

rg 'logger\.(info|error|warning|debug)\(f"' --type py

Pass criteria: Use extra dict for structured data

Severity: Medium

Recommendation: Structured logs are easier to parse, search, and aggregate

---

Category: Class Patterns

Abstract Base Classes

What to check: Use ABC for extensible component hierarchies

Good pattern (from rag/core/base.py):

from abc import ABC, abstractmethod

class Component(ABC):
    @abstractmethod
    def process(self, documents: list[Document]) -> ProcessingResult:
        pass

class Parser(Component):
    @abstractmethod
    def parse(self, source: str) -> ProcessingResult:
        pass

Severity: Medium

---

Classmethod for Factory/Service Methods

What to check: Use @classmethod for service-layer methods

Good pattern (from server/services/dataset_service.py):

class DatasetService:
    @classmethod
    def list_datasets(cls, namespace: str, project: str) -> list[Dataset]:
        project_config = ProjectService.load_config(namespace, project)
        return project_config.datasets or []

    @classmethod
    async def add_file_to_dataset(cls, ...) -> tuple[bool, MetadataFileContent]:
        ...

Severity: Medium

---

Property for Computed Attributes

What to check: Use @property for computed or derived values

Good pattern:

class Model:
    @property
    def config_tools(self) -> list[ToolDefinition]:
        return [
            ToolDefinition.from_datamodel_tool(t)
            for t in self._client._model_config.tools or []
        ]

Severity: Low

Security Best Practices Checklist

Security patterns for LlamaFarm Python components.

---

Category: Injection Prevention

No eval() or exec() with User Input

What to check: Never use eval or exec with untrusted data

Search pattern:

rg "\beval\(|\bexec\(" --type py

Pass criteria: No eval/exec calls, or only with hardcoded/trusted strings

Severity: Critical

Why it matters: eval/exec can execute arbitrary code, leading to complete system compromise

---

No Shell Injection

What to check: Avoid shell=True with subprocess

Bad pattern:

import subprocess

# DANGEROUS - allows shell injection
subprocess.run(f"process {user_input}", shell=True)
os.system(f"cat {filename}")  # Also dangerous

Good pattern:

import subprocess

# Safe - no shell interpretation
subprocess.run(["process", user_input], shell=False)
subprocess.run(["cat", filename])

Search pattern:

rg "subprocess.*shell=True|os\.system\(" --type py

Pass criteria: No shell=True with variable input; no os.system()

Severity: Critical

---

SQL Injection Prevention

What to check: Use parameterized queries, never string interpolation

Bad pattern:

# DANGEROUS - SQL injection
query = f"SELECT * FROM users WHERE name = '{name}'"
cursor.execute(query)

query = "SELECT * FROM docs WHERE id = " + doc_id

Good pattern:

# Safe - parameterized query
query = "SELECT * FROM users WHERE name = :name"
cursor.execute(query, {"name": name})

# With SQLAlchemy
stmt = select(User).where(User.name == name)

Search pattern:

rg "f\".*SELECT|f'.*SELECT|\".*SELECT.*\{|'.*SELECT.*\{" --type py

Pass criteria: No string interpolation in SQL queries

Severity: Critical

---

Category: Path Traversal Prevention

Validate File Paths

What to check: Validate paths to prevent directory traversal attacks

Bad pattern:

def get_file(filename: str):
    # DANGEROUS - allows ../../../etc/passwd
    return open(f"/data/{filename}").read()

Good pattern:

from pathlib import Path

BASE_DIR = Path("/data").resolve()

def get_file(filename: str) -> str:
    # Resolve the full path
    requested_path = (BASE_DIR / filename).resolve()

    # Verify it's within allowed directory
    if not requested_path.is_relative_to(BASE_DIR):
        raise ValueError(f"Invalid path: {filename}")

    if not requested_path.is_file():
        raise FileNotFoundError(f"File not found: {filename}")

    return requested_path.read_text()

Search pattern:

rg "open\(.*\+|Path\(.*\+" --type py

Pass criteria: All file paths validated against base directory

Severity: Critical

---

Resolve Paths Before Use

What to check: Always resolve() paths before security checks

Good pattern:

from pathlib import Path

def safe_path(base_dir: Path, user_path: str) -> Path:
    # Resolve to absolute path (resolves .., symlinks)
    full_path = (base_dir / user_path).resolve()

    # Check containment AFTER resolution
    if not full_path.is_relative_to(base_dir.resolve()):
        raise ValueError("Path traversal detected")

    return full_path

Why resolve() matters:

• Removes .. components
• Follows symlinks
• Converts to absolute path

Severity: High

---

Symlink Protection

What to check: Consider symlink attacks in path validation

Good pattern:

def safe_read(base_dir: Path, filename: str) -> str:
    path = (base_dir / filename).resolve()

    # Check containment
    if not path.is_relative_to(base_dir.resolve()):
        raise ValueError("Path outside allowed directory")

    # Optionally reject symlinks
    if path.is_symlink():
        raise ValueError("Symlinks not allowed")

    return path.read_text()

Severity: Medium

---

Category: Secrets Management

No Hardcoded Secrets

What to check: No secrets in source code

Search pattern:

rg -i "(api_key|apikey|password|secret|token|credential)\s*=\s*['\"][^'\"]+['\"]" --type py

Pass criteria: All secrets loaded from environment variables or secret managers

Severity: Critical

---

Environment Variables for Secrets

What to check: Use pydantic-settings for configuration

Good pattern (from server/core/settings.py):

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    # Secrets from environment
    DATABASE_URL: str
    API_KEY: str
    JWT_SECRET: str

    # Optional secrets with defaults only for non-sensitive values
    LOG_LEVEL: str = "INFO"

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
    )

settings = Settings()

Severity: High

---

No Secrets in Logs

What to check: Never log sensitive data

Bad pattern:

logger.info(f"Connecting with token: {api_token}")
logger.debug(f"User password: {password}")

Good pattern:

logger.info("Connecting to API", extra={"token_prefix": api_token[:8] + "..."})
logger.debug("User authentication attempt", extra={"user_id": user_id})

Search pattern:

rg -i "logger?\.(info|debug|warning|error).*password|token|secret|key" --type py

Pass criteria: No secrets in log statements

Severity: Critical

---

Secrets in Error Messages

What to check: Don't expose secrets in exceptions

Bad pattern:

raise AuthError(f"Invalid token: {token}")

Good pattern:

raise AuthError("Invalid or expired authentication token")

Severity: High

---

Category: Input Validation

Pydantic for All External Input

What to check: Use Pydantic models for request validation

Good pattern:

from pydantic import BaseModel, Field, EmailStr

class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str = Field(min_length=1, max_length=100)
    age: int = Field(ge=0, le=150)

@app.post("/users")
async def create_user(request: CreateUserRequest):
    # request is already validated
    return await user_service.create(request)

Severity: High

---

Validate at API Boundaries

What to check: Validate input as early as possible

Good pattern:

@app.post("/documents")
async def create_document(
    request: DocumentRequest,  # Validated by Pydantic
    current_user: User = Depends(get_current_user),  # Authenticated
):
    # By this point, input is validated and user is authenticated
    return await document_service.create(request, current_user)

Severity: Medium

---

Sanitize for Output Context

What to check: Escape output appropriately for context

Good pattern:

import html

# For HTML output
safe_content = html.escape(user_content)

# For JSON (Pydantic handles this)
return response.model_dump_json()

Severity: Medium (for web responses)

---

Category: Deserialization Safety

Safe YAML Loading

What to check: Always use safe_load for YAML

Bad pattern:

import yaml

# DANGEROUS - allows arbitrary Python object instantiation
data = yaml.load(user_data)  # Full loader
data = yaml.load(user_data, Loader=yaml.FullLoader)  # Still dangerous

Good pattern:

import yaml

# Safe - only allows basic types
data = yaml.safe_load(user_data)

# Or use ruamel.yaml with safe settings
from ruamel.yaml import YAML
yaml = YAML(typ='safe')

Search pattern:

rg "yaml\.load\(" --type py | rg -v "safe_load"

Pass criteria: Only safe_load used for untrusted YAML

Severity: Critical

---

Avoid Pickle with Untrusted Data

What to check: Never unpickle untrusted data

Search pattern:

rg "pickle\.load|pickle\.loads" --type py

Pass criteria: No pickle with user-provided data; use JSON instead

Severity: Critical

Why it matters: Pickle can execute arbitrary code during deserialization

---

JSON is Safe (but validate structure)

What to check: JSON is safe to parse, but validate the structure

Good pattern:

import json
from pydantic import BaseModel

# Parse JSON (safe)
data = json.loads(user_input)

# Validate structure with Pydantic
validated = MyModel.model_validate(data)

Severity: Low

---

Category: Authentication & Authorization

Token Validation

What to check: Properly validate authentication tokens

Good pattern:

import jwt
from fastapi import HTTPException, Depends
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    try:
        payload = jwt.decode(
            token,
            settings.JWT_SECRET,
            algorithms=["HS256"]
        )
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = await get_user(user_id)
    if user is None:
        raise HTTPException(status_code=401, detail="User not found")
    return user

Severity: High

---

Dependency Injection for Auth

What to check: Use FastAPI dependencies for authentication

Good pattern:

from fastapi import Depends

@app.get("/protected")
async def protected_route(
    current_user: User = Depends(get_current_user)
):
    return {"user": current_user.email}

@app.delete("/admin/users/{user_id}")
async def delete_user(
    user_id: str,
    current_user: User = Depends(require_admin),  # Admin check
):
    return await user_service.delete(user_id)

Pass criteria: All protected routes use auth dependencies

Severity: High

---

Authorization Checks

What to check: Verify user has permission for the action

Good pattern:

async def get_document(
    document_id: str,
    current_user: User = Depends(get_current_user),
):
    document = await document_service.get(document_id)
    if document is None:
        raise HTTPException(status_code=404)

    # Authorization check
    if document.owner_id != current_user.id and not current_user.is_admin:
        raise HTTPException(status_code=403, detail="Access denied")

    return document

Severity: High

---

Category: Rate Limiting

API Rate Limiting

What to check: Implement rate limiting for public endpoints

Good pattern:

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.get("/api/public")
@limiter.limit("10/minute")
async def public_endpoint(request: Request):
    ...

@app.post("/api/login")
@limiter.limit("5/minute")
async def login(request: Request):
    ...

Severity: Medium

---

Resource Limits

What to check: Limit resource consumption

Good pattern:

from pydantic import BaseModel, Field

class UploadRequest(BaseModel):
    # Limit file size
    content: str = Field(max_length=10_000_000)  # 10MB

class BatchRequest(BaseModel):
    # Limit batch size
    items: list[str] = Field(max_length=100)

# Timeout for long operations
result = await asyncio.wait_for(
    long_operation(),
    timeout=30.0
)

Severity: Medium

---

Category: Dependency Security

Pin Dependencies

What to check: Pin exact versions in production

Good pattern (pyproject.toml):

[project]
dependencies = [
    "fastapi>=0.100.0,<0.200.0",
    "pydantic>=2.0.0,<3.0.0",
]

# Or use uv.lock for exact pinning

Severity: Medium

---

Audit Dependencies

What to check: Regularly check for vulnerabilities

Commands:

# With pip-audit
pip-audit

# With safety
safety check

# With uv
uv pip audit

Severity: Medium

---

Category: Secure Defaults

HTTPS in Production

What to check: Enforce HTTPS for production APIs

Good pattern:

from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

if settings.ENVIRONMENT == "production":
    app.add_middleware(HTTPSRedirectMiddleware)

Severity: High

---

Secure Headers

What to check: Set security headers

Good pattern:

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        return response

app.add_middleware(SecurityHeadersMiddleware)

Severity: Medium

---

CORS Configuration

What to check: Configure CORS appropriately

Good pattern:

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.ALLOWED_ORIGINS,  # Not ["*"] in production
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

Bad pattern:

# DANGEROUS in production
allow_origins=["*"]

Severity: Medium

Pytest Testing Checklist

Testing patterns for LlamaFarm Python components.

---

Category: Test Organization

Test File Naming and Structure

What to check: Tests follow naming conventions and mirror source structure

Pass criteria:

• Test files named test_*.py or *_test.py
• Tests in tests/ directory at component root
• Mirror source structure: src/services/ -> tests/services/

Example (from rag/):

rag/
├── core/
│   └── base.py
├── tasks/
│   └── ingest_tasks.py
└── tests/
    ├── conftest.py
    ├── test_base.py
    └── test_ingest_tasks.py

Severity: Low

---

Test Function Naming

What to check: Test function names describe behavior being tested

Good pattern:

def test_process_document_returns_chunks():
    ...

def test_process_document_raises_on_empty_input():
    ...

def test_embedder_handles_connection_failure():
    ...

Bad pattern:

def test_1():
    ...

def test_process():  # Too vague
    ...

Severity: Low

---

conftest.py for Shared Fixtures

What to check: Shared fixtures in conftest.py

Good pattern (from rag/tests/conftest.py):

"""Essential pytest configuration and fixtures."""

import sys
from pathlib import Path

import pytest

# Add parent directories to path
rag_dir = Path(__file__).parent.parent
sys.path.insert(0, str(rag_dir))

from core.base import Document


@pytest.fixture
def sample_documents() -> list[Document]:
    """Create sample documents for testing."""
    return [
        Document(
            id="doc1",
            content="This is a test document",
            metadata={"type": "technical"},
            embeddings=[0.1, 0.2, 0.3],
        ),
    ]

Severity: Medium

---

Category: Fixtures

Fixture Scope Selection

What to check: Use appropriate fixture scope

Good pattern:

@pytest.fixture(scope="session")
def database_connection():
    """Expensive setup - share across all tests in session."""
    conn = create_connection()
    yield conn
    conn.close()

@pytest.fixture(scope="module")
def model_instance():
    """Share within test module."""
    return load_model()

@pytest.fixture  # scope="function" is default
def clean_state():
    """Fresh state for each test."""
    state = create_state()
    yield state
    state.cleanup()

Pass criteria:

• scope="session" for expensive resources (DB connections, model loading)
• scope="module" for module-level shared resources
• scope="function" (default) for test isolation

Severity: Medium

---

Temporary Directory Fixtures

What to check: Use temp directories for file-based tests

Good pattern (from rag/tests/conftest.py):

from collections.abc import Generator

@pytest.fixture
def temp_dir() -> Generator[str, None, None]:
    """Create a temporary directory for tests."""
    temp_dir = tempfile.mkdtemp()
    yield temp_dir
    shutil.rmtree(temp_dir, ignore_errors=True)

@pytest.fixture
def sample_csv_file(temp_dir: str) -> str:
    """Create a temporary CSV file with sample data."""
    csv_path = Path(temp_dir) / "test.csv"
    csv_path.write_text("name,value\ntest,123")
    return str(csv_path)

Severity: Medium

---

Factory Fixtures

What to check: Use factories for customizable test data

Good pattern:

@pytest.fixture
def make_document():
    """Factory for creating test documents."""
    def _make(
        content: str = "test content",
        id: str | None = None,
        **metadata
    ) -> Document:
        return Document(
            content=content,
            id=id or str(uuid.uuid4()),
            metadata=metadata
        )
    return _make

def test_process_with_metadata(make_document):
    doc = make_document(content="custom", type="article", author="test")
    result = process(doc)
    assert result.metadata["type"] == "article"

Severity: Low

---

Autouse Fixtures Sparingly

What to check: Limit autouse to truly universal setup

Search pattern:

rg "@pytest.fixture\(.*autouse=True" --type py tests/

Pass criteria: autouse only for:

• Environment variable reset
• Logging configuration
• Global state cleanup

Bad pattern:

@pytest.fixture(autouse=True)
def setup_database():  # Too heavy for autouse
    ...

Severity: Low

---

Category: Async Testing

pytest-asyncio Markers

What to check: Async tests are properly marked

Good pattern:

import pytest

@pytest.mark.asyncio
async def test_async_operation():
    result = await some_async_function()
    assert result is not None

@pytest.mark.asyncio
async def test_model_loading():
    model = await load_model("test-model")
    assert model.model_id == "test-model"

Search pattern (find unmarked async tests):

rg "async def test_" --type py tests/ -B 2 | rg -v "@pytest.mark.asyncio"

Pass criteria: All async test functions have @pytest.mark.asyncio

Severity: High (tests fail without marker)

---

Async Fixtures

What to check: Async fixtures properly defined

Good pattern:

@pytest.fixture
async def async_client():
    """Create async HTTP client for testing."""
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        yield client

@pytest.fixture
async def loaded_model():
    """Load model for testing."""
    model = LanguageModel("test-model", "cpu")
    await model.load()
    yield model
    await model.unload()

Severity: Medium

---

pytest-asyncio Configuration

What to check: Configure asyncio mode in pyproject.toml

Good pattern:

[tool.pytest.ini_options]
asyncio_mode = "auto"  # or "strict"
asyncio_default_fixture_loop_scope = "function"

Severity: Medium

---

Category: Mocking

Mock at the Right Level

What to check: Mock where imported, not where defined

Good pattern:

# If service.py does: from external import api_client
# Mock in service's namespace, not external's
@patch("myapp.service.api_client")
def test_service_calls_api(mock_client):
    mock_client.get.return_value = {"data": "test"}
    result = service.fetch_data()
    assert result == {"data": "test"}

Bad pattern:

@patch("external.api_client")  # Wrong - mock at definition site
def test_service(mock_client):
    ...

Severity: High

---

Mock Return Values and Side Effects

What to check: Set appropriate mock behaviors

Good pattern:

from unittest.mock import Mock, AsyncMock

# Sync mock with return value
mock_client.get.return_value = Mock(
    status_code=200,
    json=Mock(return_value={"key": "value"})
)

# Async mock
mock_client.fetch = AsyncMock(return_value={"data": "test"})

# Side effects for multiple calls
mock_client.get.side_effect = [
    {"first": "call"},
    {"second": "call"},
    ConnectionError("Failed"),
]

Severity: Medium

---

Avoid Over-Mocking

What to check: Don't mock implementation details

Pass criteria:

• Mock external dependencies (APIs, databases, file systems)
• Don't mock internal methods being tested
• Use real objects when practical
• Integration tests with minimal mocking

Bad pattern:

def test_process_document(mocker):
    # Over-mocking - testing mock behavior, not real code
    mocker.patch.object(processor, '_validate')
    mocker.patch.object(processor, '_transform')
    mocker.patch.object(processor, '_save')
    processor.process(doc)

Severity: Medium

---

pytest-mock for Convenience

What to check: Use pytest-mock's mocker fixture

Good pattern:

def test_with_mocker(mocker):
    mock_api = mocker.patch("myapp.service.external_api")
    mock_api.return_value = {"status": "ok"}

    result = service.call_api()

    mock_api.assert_called_once()
    assert result["status"] == "ok"

Severity: Low

---

Category: Parametrization

Parametrized Tests

What to check: Use parametrize for multiple test cases

Good pattern:

@pytest.mark.parametrize("input,expected", [
    ("hello", "HELLO"),
    ("world", "WORLD"),
    ("", ""),
    ("MiXeD", "MIXED"),
])
def test_uppercase(input, expected):
    assert uppercase(input) == expected

@pytest.mark.parametrize("invalid_input", [
    None,
    123,
    [],
    {},
])
def test_uppercase_rejects_invalid_types(invalid_input):
    with pytest.raises(TypeError):
        uppercase(invalid_input)

Severity: Low

---

Parametrize IDs for Readability

What to check: Add IDs for readable test output

Good pattern:

@pytest.mark.parametrize("input,expected", [
    pytest.param("hello", "HELLO", id="simple-word"),
    pytest.param("", "", id="empty-string"),
    pytest.param("a b c", "A B C", id="with-spaces"),
    pytest.param("123abc", "123ABC", id="mixed-chars"),
])
def test_uppercase(input, expected):
    assert uppercase(input) == expected

Severity: Low

---

Category: Assertions

Specific Assertions

What to check: Use specific assertion patterns

Good pattern:

# Equality
assert result == expected
assert result != other

# Membership
assert item in collection
assert "key" in dictionary

# Type checking
assert isinstance(result, Document)

# Approximate equality for floats
assert result == pytest.approx(3.14159, rel=1e-3)

Bad pattern:

assert result  # Not specific - what value is expected?
assert bool(result)  # Same problem

Severity: Low

---

Exception Testing

What to check: Test expected exceptions with context

Good pattern:

def test_raises_on_invalid_input():
    with pytest.raises(ValueError, match="must be positive"):
        process(-1)

def test_raises_specific_exception():
    with pytest.raises(ConfigError) as exc_info:
        load_config("nonexistent.yaml")
    assert "not found" in str(exc_info.value)
    assert exc_info.value.path == "nonexistent.yaml"

Severity: Medium

---

Assert Message Context

What to check: Add context to assertions when helpful

Good pattern:

def test_document_processing():
    result = process(documents)

    assert len(result.documents) == 10, f"Expected 10 docs, got {len(result.documents)}"
    assert result.errors == [], f"Unexpected errors: {result.errors}"

Severity: Low

---

Category: Test Markers

Custom Markers for Test Categories

What to check: Define and use custom markers

Good pattern (in conftest.py):

def pytest_configure(config):
    """Configure pytest markers."""
    config.addinivalue_line("markers", "integration: integration tests")
    config.addinivalue_line("markers", "slow: slow tests")
    config.addinivalue_line("markers", "gpu: tests requiring GPU")

Usage:

@pytest.mark.integration
def test_database_connection():
    ...

@pytest.mark.slow
def test_full_pipeline():
    ...

# Run only fast tests
# pytest -m "not slow"

# Run integration tests
# pytest -m integration

Severity: Low

---

Skip Markers

What to check: Use skip markers appropriately

Good pattern:

import pytest

@pytest.mark.skip(reason="Feature not implemented yet")
def test_future_feature():
    ...

@pytest.mark.skipif(
    not torch.cuda.is_available(),
    reason="CUDA not available"
)
def test_gpu_inference():
    ...

@pytest.mark.xfail(reason="Known bug, fix pending")
def test_known_issue():
    ...

Severity: Low

---

Category: Test Coverage

Coverage Configuration

What to check: Coverage properly configured

Good pattern (pyproject.toml):

[tool.pytest.ini_options]
addopts = "--cov=src --cov-report=term-missing --cov-report=html"
testpaths = ["tests"]

[tool.coverage.run]
source = ["src"]
omit = ["*/tests/*", "*/__pycache__/*"]
branch = true

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "if TYPE_CHECKING:",
    "raise NotImplementedError",
]

Severity: Low

---

Critical Path Coverage

What to check: Core functionality is tested

Pass criteria:

• All public API functions have tests
• Error paths are tested
• Edge cases covered
• Integration points tested

Severity: High

Type Hints and Pydantic Checklist

Type annotation patterns for LlamaFarm Python components.

---

Category: Basic Type Hints

Function Return Types

What to check: All public functions should have return type hints

Good pattern:

def process_document(
    content: str,
    metadata: dict[str, Any] | None = None,
) -> ProcessingResult:
    ...

async def load_model(model_id: str) -> BaseModel:
    ...

Search pattern:

rg "^def [a-z_]+\([^)]*\):" --type py | rg -v " -> "

Pass criteria: All public functions have return type annotations

Severity: Medium

---

None Return Type

What to check: Functions returning None should be explicit

Good pattern:

def log_event(event: str) -> None:
    logger.info(event)

async def cleanup() -> None:
    await close_connections()

Severity: Low

---

Category: Modern Type Syntax (Python 3.10+)

Union Types with Pipe Operator

What to check: Use | syntax for unions (PEP 604)

Good: str | None, int | float Bad: Optional[str], Union[int, float]

Search pattern:

rg "Optional\[|Union\[" --type py

Pass criteria: Use X | Y syntax

Severity: Low

Recommendation: Run ruff check --fix --select UP to auto-upgrade

---

Built-in Generic Types (PEP 585)

What to check: Use lowercase built-in generics

Good:

items: list[str]
mapping: dict[str, int]
coordinates: tuple[float, float]
unique_ids: set[int]

Bad:

from typing import List, Dict, Tuple, Set
items: List[str]  # DEPRECATED

Search pattern:

rg "from typing import.*(List|Dict|Tuple|Set)" --type py

Pass criteria: Use lowercase built-in generics

Severity: Low

---

collections.abc for Abstract Types

What to check: Use collections.abc for abstract types

Good pattern:

from collections.abc import Sequence, Mapping, Iterator, Callable, AsyncGenerator

def process(items: Sequence[str]) -> Iterator[str]:
    for item in items:
        yield item.upper()

async def stream_data() -> AsyncGenerator[bytes]:
    ...

Bad pattern:

from typing import Sequence, Iterator  # DEPRECATED in 3.9+

Severity: Low

---

Category: Variable Annotations

Complex Variable Types

What to check: Complex variables should have type annotations

Good pattern:

# Module-level
_models: dict[str, BaseModel] = {}
_cleanup_task: asyncio.Task[None] | None = None

# Inside functions
results: list[Document] = []
cache: dict[str, Model] = {}

Severity: Low

---

Final for Constants

What to check: Use Final for constants

Good pattern:

from typing import Final

MAX_RETRIES: Final = 3
DEFAULT_TIMEOUT: Final[float] = 30.0
API_VERSION: Final = "v1"

Severity: Low

---

Category: Generic Types

TypeVar for Generic Functions

What to check: Use TypeVar for type-preserving generics

Good pattern:

from typing import TypeVar

T = TypeVar("T")

def first(items: Sequence[T]) -> T | None:
    return items[0] if items else None

# With bounds
ModelT = TypeVar("ModelT", bound=BaseModel)

def load(model_class: type[ModelT], data: dict) -> ModelT:
    return model_class(**data)

Severity: Medium

---

Protocol for Structural Typing

What to check: Use Protocol for duck typing interfaces

Good pattern (from rag/core/base.py pattern):

from typing import Protocol

class Processable(Protocol):
    def process(self, documents: list[Document]) -> ProcessingResult: ...

class Embeddable(Protocol):
    def embed(self, texts: list[str]) -> list[list[float]]: ...
    def get_embedding_dimension(self) -> int: ...

def run_processor(p: Processable) -> ProcessingResult:
    return p.process([])

Search pattern:

rg "class \w+\(Protocol\)" --type py

Severity: Medium

---

TypedDict for Dict Structures

What to check: Use TypedDict for known dict structures

Good pattern:

from typing import TypedDict, NotRequired

class DocumentMetadata(TypedDict):
    source: str
    page: int
    author: NotRequired[str]
    timestamp: NotRequired[float]

def process_doc(meta: DocumentMetadata) -> None:
    print(meta["source"])  # Type-safe access

Severity: Medium

---

Category: Callable Types

Callable Signatures

What to check: Use Callable for function parameters

Good pattern:

from collections.abc import Callable

def apply(
    items: list[T],
    transform: Callable[[T], R],
) -> list[R]:
    return [transform(item) for item in items]

# For async callables
async def with_retry(
    func: Callable[[], Awaitable[T]],
    max_retries: int = 3,
) -> T:
    ...

Severity: Medium

---

ParamSpec for Decorators

What to check: Use ParamSpec for type-safe decorators

Good pattern:

from typing import ParamSpec, TypeVar
from functools import wraps

P = ParamSpec("P")
R = TypeVar("R")

def logged(func: Callable[P, R]) -> Callable[P, R]:
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        logger.info(f"Calling {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

Severity: Low

---

Category: Pydantic Models

Pydantic v2 Field Syntax

What to check: Use Pydantic v2 Field patterns

Good pattern:

from pydantic import BaseModel, Field, ConfigDict

class EmbeddingRequest(BaseModel):
    model: str
    input: str | list[str]
    encoding_format: Literal["float", "base64"] = "float"

    model_config = ConfigDict(str_strip_whitespace=True)

Severity: Medium

---

Annotated with Field Constraints

What to check: Use Annotated for complex field constraints

Good pattern:

from typing import Annotated
from pydantic import Field

class Model(BaseModel):
    name: Annotated[str, Field(min_length=1, max_length=100)]
    count: Annotated[int, Field(ge=0, le=1000)]
    ratio: Annotated[float, Field(ge=0.0, le=1.0)]

Severity: Medium

---

Field Validators (Pydantic v2)

What to check: Use v2 validator decorators

Good pattern:

from pydantic import field_validator, model_validator

class Config(BaseModel):
    url: str
    port: int

    @field_validator("url")
    @classmethod
    def validate_url(cls, v: str) -> str:
        if not v.startswith(("http://", "https://")):
            raise ValueError("URL must start with http:// or https://")
        return v

    @model_validator(mode="after")
    def validate_config(self) -> "Config":
        # Cross-field validation
        return self

Bad pattern:

@validator("url")  # DEPRECATED - Pydantic v1
@root_validator  # DEPRECATED - Pydantic v1

Search pattern:

rg "@validator|@root_validator" --type py

Severity: Medium

---

Generic Pydantic Models

What to check: Use Generic for reusable response models

Good pattern:

from typing import Generic, TypeVar
from pydantic import BaseModel

T = TypeVar("T")

class Response(BaseModel, Generic[T]):
    data: T
    success: bool = True
    error: str | None = None

class PaginatedResponse(BaseModel, Generic[T]):
    items: list[T]
    total: int
    page: int
    per_page: int

# Usage
Response[User](data=user, success=True)
PaginatedResponse[Document](items=docs, total=100, page=1, per_page=10)

Severity: Low

---

Category: Type Narrowing

Type Guards

What to check: Use TypeGuard for runtime type narrowing

Good pattern:

from typing import TypeGuard

def is_string_list(val: list[Any]) -> TypeGuard[list[str]]:
    return all(isinstance(x, str) for x in val)

def process(items: list[Any]) -> None:
    if is_string_list(items):
        # items is now list[str]
        for s in items:
            print(s.upper())

Severity: Low

---

isinstance Narrowing

What to check: Use isinstance for type narrowing

Good pattern:

def process(value: str | int | None) -> str:
    if value is None:
        return "default"
    if isinstance(value, int):
        return str(value)
    # value is now str
    return value.upper()

Severity: Low

---

Category: Async Types

AsyncGenerator Type

What to check: Async generators have proper return types

Good pattern (from server/agents/base/agent.py):

from collections.abc import AsyncGenerator

async def run_async_stream(
    self,
    messages: list[LFChatCompletionMessageParam] | None = None,
) -> AsyncGenerator[LFChatCompletionChunk]:
    async for chunk in self._client.stream_chat(messages=messages):
        yield chunk

Severity: Medium

---

Awaitable and Coroutine Types

What to check: Use proper async type hints

Good pattern:

from collections.abc import Awaitable, Coroutine

async def fetch_data() -> dict[str, Any]:
    ...

# Type for a coroutine parameter
def schedule(coro: Awaitable[T]) -> asyncio.Task[T]:
    return asyncio.create_task(coro)

Severity: Low

---

Category: Type Aliases

Type Aliases for Complex Types

What to check: Create type aliases for complex or repeated types

Good pattern:

from typing import TypeAlias

# Simple alias
JsonDict: TypeAlias = dict[str, Any]
Embedding: TypeAlias = list[float]

# Complex alias
MessageHistory: TypeAlias = list[dict[str, str | list[dict[str, str]]]]
Handler: TypeAlias = Callable[[Request], Awaitable[Response]]

Severity: Low

---

NewType for Distinct Types

What to check: Use NewType for distinct string/int types

Good pattern:

from typing import NewType

UserId = NewType("UserId", str)
DocumentId = NewType("DocumentId", str)

def get_user(user_id: UserId) -> User:
    ...

# Usage
user_id = UserId("user-123")
get_user(user_id)  # OK
get_user("user-123")  # Type error - need explicit NewType wrapper

Severity: Low

---

Category: Literal Types

Literal for String Enums

What to check: Use Literal for fixed string values

Good pattern:

from typing import Literal

LogLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR"]
Status = Literal["pending", "running", "completed", "failed"]

def set_log_level(level: LogLevel) -> None:
    ...

class Task(BaseModel):
    status: Status = "pending"

Severity: Medium