Project Architecture Patterns

Purpose

Establish consistent architecture patterns for the rRNA-Phylo project, covering backend design, service organization, API conventions, and testing strategies.

When to Use

This skill activates when:

•Setting up new services or modules
•Designing API endpoints
•Implementing background tasks
•Working with configuration or dependencies
•Writing tests
•Integrating external tools (HMMER, BLAST, alignment tools)

Tech Stack

Core Backend

•FastAPI: Modern async web framework
•Pydantic: Data validation and settings
•SQLAlchemy: ORM for database operations
•Celery: Distributed task queue for long-running jobs
•Redis: Message broker and cache
•PostgreSQL: Primary database (SQLite for dev)

Scientific Computing

•Biopython: Sequence parsing and analysis
•NumPy/Pandas: Numerical computing and data manipulation
•scikit-learn: ML baseline models
•PyTorch: Deep learning (Phase 2+)

Testing & Quality

•pytest: Testing framework
•pytest-asyncio: Async test support
•pytest-cov: Coverage reporting
•httpx: Async HTTP client for API tests
•ruff: Fast linter
•black: Code formatter

Project Structure

code

backend/
├── app/
│   ├── __init__.py
│   ├── main.py                    # FastAPI application
│   ├── config.py                  # Settings (via Pydantic)
│   │
│   ├── models/                    # SQLAlchemy ORM models
│   │   ├── __init__.py
│   │   ├── base.py               # Base model class
│   │   ├── job.py                # Job model
│   │   ├── sequence.py           # Sequence model
│   │   └── tree.py               # Phylogenetic tree model
│   │
│   ├── schemas/                   # Pydantic schemas (API contracts)
│   │   ├── __init__.py
│   │   ├── common.py             # Shared schemas
│   │   ├── rrna.py               # rRNA detection schemas
│   │   ├── phylo.py              # Phylogenetics schemas
│   │   └── job.py                # Job status schemas
│   │
│   ├── api/                       # API routes
│   │   ├── __init__.py
│   │   ├── deps.py               # Dependency injection
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── router.py         # Main router
│   │       ├── rrna.py           # rRNA endpoints
│   │       ├── phylo.py          # Phylogenetics endpoints
│   │       └── jobs.py           # Job management
│   │
│   ├── services/                  # Business logic layer
│   │   ├── rrna/                 # rRNA detection service
│   │   ├── phylo/                # Phylogenetics service
│   │   └── sequences/            # Sequence processing
│   │
│   ├── workers/                   # Celery tasks
│   │   ├── __init__.py
│   │   ├── celery_app.py         # Celery config
│   │   └── tasks.py              # Task definitions
│   │
│   ├── core/                      # Core utilities
│   │   ├── __init__.py
│   │   ├── errors.py             # Custom exceptions
│   │   └── logging.py            # Logging setup
│   │
│   └── db/                        # Database utilities
│       ├── __init__.py
│       ├── session.py            # DB session management
│       └── migrations/           # Alembic migrations
│
├── tests/
│   ├── conftest.py               # Pytest fixtures
│   ├── unit/                     # Unit tests
│   ├── integration/              # Integration tests
│   └── fixtures/                 # Test data
│
├── requirements.txt
├── pyproject.toml
└── Dockerfile

Core Patterns

1. Configuration Management

Pattern: Use Pydantic Settings for type-safe configuration.

python

# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    """Application settings."""

    # API Settings
    api_v1_prefix: str = "/api/v1"
    project_name: str = "rRNA-Phylo"
    debug: bool = False

    # Database
    database_url: str = "sqlite:///./rrna_phylo.db"

    # Celery
    celery_broker_url: str = "redis://localhost:6379/0"
    celery_result_backend: str = "redis://localhost:6379/0"

    # External Tools
    hmmer_path: str = "/usr/local/bin/hmmsearch"
    blast_path: str = "/usr/local/bin/blastn"

    # Data Paths
    hmm_profiles_dir: str = "./data/hmm_profiles"
    blast_db_dir: str = "./data/blast_dbs"

    # ML Models
    ml_models_dir: str = "./models"

    class Config:
        env_file = ".env"
        case_sensitive = False

@lru_cache()
def get_settings() -> Settings:
    """Get cached settings instance."""
    return Settings()

Usage in dependencies:

python

# app/api/deps.py
from fastapi import Depends
from app.config import Settings, get_settings

async def get_config() -> Settings:
    return get_settings()

2. Service Layer Pattern

Pattern: Encapsulate business logic in service classes.

python

# app/services/rrna/detector.py
from typing import List, Optional
from app.schemas.rrna import RRNADetectionResult, RRNAType
from app.config import Settings

class RRNADetectorService:
    """Service for detecting rRNA in sequences."""

    def __init__(self, settings: Settings):
        self.settings = settings
        self.hmm_detector = HMMDetector(settings.hmm_profiles_dir)
        self.pattern_detector = PatternDetector()

    async def detect(
        self,
        sequence: str,
        methods: List[str] = ["hmm", "pattern"],
        min_confidence: float = 0.8
    ) -> List[RRNADetectionResult]:
        """
        Detect rRNA in sequence using multiple methods.

        Args:
            sequence: DNA/RNA sequence string
            methods: Detection methods to use
            min_confidence: Minimum confidence threshold

        Returns:
            List of detection results
        """
        results = []

        if "hmm" in methods:
            hmm_results = await self.hmm_detector.detect(sequence)
            results.extend(hmm_results)

        if "pattern" in methods:
            pattern_results = await self.pattern_detector.detect(sequence)
            results.extend(pattern_results)

        # Filter by confidence
        results = [r for r in results if r.confidence >= min_confidence]

        # Deduplicate and merge
        return self._merge_results(results)

    def _merge_results(
        self,
        results: List[RRNADetectionResult]
    ) -> List[RRNADetectionResult]:
        """Merge overlapping detections."""
        # Implementation here
        pass

Usage in API:

python

# app/api/v1/rrna.py
from fastapi import APIRouter, Depends
from app.services.rrna.detector import RRNADetectorService
from app.api.deps import get_rrna_detector

router = APIRouter()

@router.post("/detect")
async def detect_rrna(
    sequence: str,
    detector: RRNADetectorService = Depends(get_rrna_detector)
):
    """Detect rRNA in sequence."""
    results = await detector.detect(sequence)
    return {"results": results}

3. Pydantic Schemas (API Contracts)

Pattern: Define clear input/output schemas with validation.

python

# app/schemas/rrna.py
from pydantic import BaseModel, Field, validator
from typing import List, Optional
from enum import Enum

class RRNAType(str, Enum):
    """Supported rRNA types."""
    SSU_16S = "16S"
    SSU_18S = "18S"
    LSU_23S = "23S"
    LSU_28S = "28S"
    FIVE_S = "5S"
    FIVE_EIGHT_S = "5.8S"

class DetectionMethod(str, Enum):
    """Detection methods."""
    HMM = "hmm"
    BLAST = "blast"
    PATTERN = "pattern"
    ML = "ml"

class RRNADetectionRequest(BaseModel):
    """Request schema for rRNA detection."""

    sequence: str = Field(..., min_length=100, max_length=100000)
    methods: List[DetectionMethod] = Field(
        default=[DetectionMethod.HMM, DetectionMethod.PATTERN]
    )
    min_confidence: float = Field(default=0.8, ge=0.0, le=1.0)

    @validator("sequence")
    def validate_sequence(cls, v):
        """Ensure sequence contains only valid nucleotides."""
        valid_chars = set("ACGTUNacgtun-")
        if not set(v).issubset(valid_chars):
            raise ValueError("Sequence contains invalid characters")
        return v.upper()

class ConservedRegion(BaseModel):
    """Conserved region within detected rRNA."""
    name: str
    start: int
    end: int
    sequence: str
    confidence: float

class RRNADetectionResult(BaseModel):
    """Result schema for rRNA detection."""

    rrna_type: RRNAType
    start: int
    end: int
    length: int
    confidence: float
    method: DetectionMethod
    score: float

    # Quality metrics
    completeness: float = Field(..., ge=0.0, le=1.0)
    quality: str = Field(..., pattern="^(high|medium|low|very_low)$")

    # Optional details
    conserved_regions: Optional[List[ConservedRegion]] = None
    secondary_structure: Optional[str] = None

    class Config:
        json_schema_extra = {
            "example": {
                "rrna_type": "16S",
                "start": 0,
                "end": 1542,
                "length": 1542,
                "confidence": 0.95,
                "method": "hmm",
                "score": 1250.5,
                "completeness": 0.98,
                "quality": "high"
            }
        }

4. Dependency Injection

Pattern: Use FastAPI's dependency injection for services.

python

# app/api/deps.py
from typing import Generator
from fastapi import Depends
from sqlalchemy.orm import Session

from app.db.session import SessionLocal
from app.config import Settings, get_settings
from app.services.rrna.detector import RRNADetectorService

# Database dependency
def get_db() -> Generator:
    """Get database session."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Service dependencies
def get_rrna_detector(
    settings: Settings = Depends(get_settings)
) -> RRNADetectorService:
    """Get rRNA detector service."""
    return RRNADetectorService(settings)

def get_phylo_service(
    settings: Settings = Depends(get_settings),
    db: Session = Depends(get_db)
):
    """Get phylogenetic analysis service."""
    from app.services.phylo.tree_builder import PhyloService
    return PhyloService(settings, db)

5. Error Handling

Pattern: Use custom exceptions and FastAPI exception handlers.

python

# app/core/errors.py
class RRNAPhyloException(Exception):
    """Base exception for rRNA-Phylo."""
    pass

class SequenceValidationError(RRNAPhyloException):
    """Raised when sequence validation fails."""
    pass

class DetectionError(RRNAPhyloException):
    """Raised when rRNA detection fails."""
    pass

class AlignmentError(RRNAPhyloException):
    """Raised when sequence alignment fails."""
    pass

class TreeBuildingError(RRNAPhyloException):
    """Raised when tree building fails."""
    pass

# app/main.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from app.core.errors import RRNAPhyloException

app = FastAPI()

@app.exception_handler(RRNAPhyloException)
async def rrna_phylo_exception_handler(
    request: Request,
    exc: RRNAPhyloException
):
    """Handle custom exceptions."""
    return JSONResponse(
        status_code=400,
        content={
            "error": exc.__class__.__name__,
            "message": str(exc)
        }
    )

6. Async Task Processing (Celery)

Pattern: Use Celery for long-running tasks.

python

# app/workers/celery_app.py
from celery import Celery
from app.config import get_settings

settings = get_settings()

celery_app = Celery(
    "rrna_phylo",
    broker=settings.celery_broker_url,
    backend=settings.celery_result_backend
)

celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_track_started=True,
    task_time_limit=3600,  # 1 hour max
)

# app/workers/tasks.py
from app.workers.celery_app import celery_app
from app.services.phylo.tree_builder import PhyloService

@celery_app.task(bind=True)
def build_phylogenetic_tree(
    self,
    sequences: List[dict],
    method: str,
    parameters: dict
):
    """
    Build phylogenetic tree (long-running task).

    Args:
        self: Task instance (for progress updates)
        sequences: List of sequences
        method: Tree building method
        parameters: Method parameters
    """
    try:
        # Update progress
        self.update_state(state="PROGRESS", meta={"stage": "alignment"})

        service = PhyloService()

        # Align sequences
        alignment = service.align_sequences(sequences)

        self.update_state(state="PROGRESS", meta={"stage": "tree_building"})

        # Build tree
        tree = service.build_tree(alignment, method, parameters)

        return {"tree": tree, "status": "completed"}

    except Exception as e:
        self.update_state(state="FAILURE", meta={"error": str(e)})
        raise

API integration:

python

# app/api/v1/phylo.py
from fastapi import APIRouter, BackgroundTasks
from app.workers.tasks import build_phylogenetic_tree

router = APIRouter()

@router.post("/tree")
async def create_tree(request: TreeBuildRequest):
    """Submit tree building job."""

    # Submit Celery task
    task = build_phylogenetic_tree.delay(
        sequences=request.sequences,
        method=request.method,
        parameters=request.parameters
    )

    return {
        "job_id": task.id,
        "status": "submitted"
    }

@router.get("/tree/{job_id}")
async def get_tree_status(job_id: str):
    """Get tree building job status."""
    task = build_phylogenetic_tree.AsyncResult(job_id)

    return {
        "job_id": job_id,
        "status": task.state,
        "result": task.result if task.ready() else None
    }

Testing Patterns

1. Test Structure

python

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.main import app
from app.db.session import Base, get_db

# Test database
SQLALCHEMY_TEST_URL = "sqlite:///./test.db"

@pytest.fixture(scope="session")
def test_engine():
    """Create test database engine."""
    engine = create_engine(SQLALCHEMY_TEST_URL)
    Base.metadata.create_all(bind=engine)
    yield engine
    Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def test_db(test_engine):
    """Create test database session."""
    TestSessionLocal = sessionmaker(bind=test_engine)
    db = TestSessionLocal()
    try:
        yield db
    finally:
        db.close()

@pytest.fixture
def client(test_db):
    """Create test client."""
    def override_get_db():
        yield test_db

    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    app.dependency_overrides.clear()

@pytest.fixture
def sample_16s_sequence():
    """Sample 16S rRNA sequence for testing."""
    return "AGAGTTTGATCCTGGCTCAG..."  # Truncated

2. Unit Test Example

python

# tests/unit/test_rrna_detector.py
import pytest
from app.services.rrna.detector import RRNADetectorService

@pytest.mark.asyncio
async def test_detect_16s_rrna(sample_16s_sequence, get_settings):
    """Test 16S rRNA detection."""
    detector = RRNADetectorService(get_settings())

    results = await detector.detect(sample_16s_sequence)

    assert len(results) > 0
    assert results[0].rrna_type == "16S"
    assert results[0].confidence >= 0.8

3. Integration Test Example

python

# tests/integration/test_api.py
def test_detect_rrna_endpoint(client, sample_16s_sequence):
    """Test rRNA detection API endpoint."""
    response = client.post(
        "/api/v1/rrna/detect",
        json={
            "sequence": sample_16s_sequence,
            "methods": ["hmm", "pattern"],
            "min_confidence": 0.8
        }
    )

    assert response.status_code == 200
    data = response.json()
    assert "results" in data
    assert len(data["results"]) > 0

API Design Conventions

1. RESTful Endpoints

code

# rRNA Detection
POST   /api/v1/rrna/detect           # Detect rRNA
GET    /api/v1/rrna/types             # List supported types

# Phylogenetics
POST   /api/v1/phylo/align            # Align sequences
POST   /api/v1/phylo/tree             # Build tree
POST   /api/v1/phylo/bootstrap        # Bootstrap analysis
POST   /api/v1/phylo/consensus        # Consensus tree

# Jobs
GET    /api/v1/jobs                   # List jobs
GET    /api/v1/jobs/{id}              # Get job status
DELETE /api/v1/jobs/{id}              # Cancel job

# Health
GET    /health                        # Health check
GET    /metrics                       # Prometheus metrics

2. Response Format

python

# Success response
{
    "success": true,
    "data": { ... },
    "metadata": {
        "timestamp": "2025-11-20T12:00:00Z",
        "version": "1.0.0"
    }
}

# Error response
{
    "success": false,
    "error": {
        "code": "VALIDATION_ERROR",
        "message": "Invalid sequence format",
        "details": { ... }
    }
}

Best Practices

✅ DO

•Use type hints everywhere
•Write docstrings for public APIs
•Use Pydantic for validation
•Implement proper logging
•Write tests for new features
•Use async/await for I/O operations
•Separate concerns (routes, services, models)
•Use dependency injection
•Handle errors gracefully
•Document API with examples

❌ DON'T

•Mix business logic with API routes
•Use synchronous I/O in async functions
•Hardcode configuration values
•Skip input validation
•Ignore exceptions
•Write god classes
•Couple services tightly
•Skip tests
•Use global state
•Block the event loop

External Tool Integration Pattern

python

# app/services/rrna/hmm.py
import subprocess
from pathlib import Path
from typing import Optional

class HMMDetector:
    """HMM-based rRNA detection using HMMER."""

    def __init__(self, profiles_dir: str, hmmsearch_path: str = "hmmsearch"):
        self.profiles_dir = Path(profiles_dir)
        self.hmmsearch_path = hmmsearch_path

    async def detect(self, sequence: str, rrna_type: str) -> dict:
        """Run HMMER search."""

        # Create temp files
        with tempfile.NamedTemporaryFile(mode='w', suffix='.fasta') as seq_file:
            seq_file.write(f">query\n{sequence}\n")
            seq_file.flush()

            profile = self.profiles_dir / f"{rrna_type}.hmm"

            # Run HMMER
            result = subprocess.run(
                [
                    self.hmmsearch_path,
                    "--tblout", "/dev/stdout",
                    str(profile),
                    seq_file.name
                ],
                capture_output=True,
                text=True,
                timeout=300
            )

            if result.returncode != 0:
                raise DetectionError(f"HMMER failed: {result.stderr}")

            # Parse output
            return self._parse_hmmer_output(result.stdout)

Related Skills: rRNA-prediction-patterns, ml-integration-patterns

Line Count: < 500 lines ✅