v2.0 Update

Type-Safe Code Patterns (v2.0)

Production-ready type safety, repository pattern, and unified logging for PMDA compliance.

Status: Production Ready

Core infrastructure complete. All new code should use these patterns. Migration guide available for existing code.

What's New in v2.0

Type Safety with Pydantic

Automatic validation and IDE autocomplete

8 Pydantic models for PERV, 91 pathogen screening, and workflows
Runtime validation catches errors before database
Auto-calculated confidence levels for detections

Repository Pattern

Database abstraction for testability

Production: RDS PostgreSQL via Data API
Testing: SQLite in-memory (10x faster, no AWS!)
Same interface, swappable backends

Unified Logging (AWS Lambda Powertools)

Structured JSON logs for PMDA audit

CloudWatch Logs Insights compatible
Correlation IDs for distributed tracing
X-Ray integration for performance analysis

CloudWatch Audit Queries

12 pre-built compliance queries

PERV detection history (CRITICAL for PMDA)
Complete run audit trail
1-minute compliance reports (vs. 1 hour manual)

Quick Start Examples

1. Using Type-Safe Models

from lib.models.pathogen import PERVTypingOutput, PERVDetectionResult

# Before (v1.0): dict with unknown structure
result = {'PERV-A': {'reads': 42, 'coverage': 0.85}}

# After (v2.0): Type-safe model with validation
result = PERVTypingOutput(
    run_id="RUN-001",
    bam_file=Path("test.bam"),
    detections={
        PERVSubtype.PERV_A: PERVDetectionResult(
            subtype=PERVSubtype.PERV_A,
            reads_aligned=42,
            coverage=0.85,  # ✅ Validated: 0.0 ≤ coverage ≤ 1.0
            mean_identity=96.5,
            confidence=PathogenConfidence.HIGH  # ✅ Auto-calculated
        )
    }
)

if result.requires_sns_alert:  # ✅ Type-safe property
    send_alert(result.to_audit_log())

2. Repository Pattern

from lib.repositories.rds_repository import RDSWorkflowRepository

# Production (RDS)
repo = RDSWorkflowRepository(
    cluster_arn=os.environ['RDS_CLUSTER_ARN'],
    secret_arn=os.environ['RDS_SECRET_ARN']
)

workflow = WorkflowExecution(run_id="RUN-001", ...)
repo.create(workflow)  # ✅ Type-safe insert

# Testing (SQLite - no AWS needed!)
from lib.repositories.sqlite_repository import SQLiteWorkflowRepository

test_repo = SQLiteWorkflowRepository(db_path=":memory:")
test_repo.create(workflow)  # ✅ Same interface!

3. Unified Logging

from lib.logging.logger import get_logger, AuditLogger

logger = get_logger("pathogen-detection")
audit = AuditLogger(service="pathogen-detection")

@logger.inject_lambda_context()
def lambda_handler(event, context):
    logger.info("Started", extra={"run_id": event['run_id']})

    # PMDA audit logging
    audit.log_perv_detection(
        run_id=event['run_id'],
        subtypes_detected=["PERV-A"],
        confidence_levels={"PERV-A": "HIGH"},
        operator_email=event['operator_email']
    )