job-forge/docs/development/coding_standards.md

Coding Standards - Job Forge

Overview

This document outlines the coding standards and best practices for the Job Forge Python/FastAPI web application. Following these standards ensures code consistency, maintainability, and quality across the project.

Python Code Style

1. PEP 8 Compliance

Job Forge follows PEP 8 with the following tools:

• Black for code formatting
• Ruff for linting and import sorting
• mypy for type checking

2. Code Formatting with Black

# Install black
pip install black

# Format all Python files
black .

# Check formatting without making changes
black --check .

# Format specific file
black app/main.py

Black Configuration (.pyproject.toml)

[tool.black]
line-length = 88
target-version = ['py312']
include = '\.pyi?$'
extend-exclude = '''
/(
  # directories
  \.eggs
  | \.git
  | \.hg
  | \.mypy_cache
  | \.tox
  | \.venv
  | build
  | dist
)/
'''

3. Linting with Ruff

# Install ruff
pip install ruff

# Lint all files
ruff check .

# Fix auto-fixable issues
ruff check --fix .

# Check specific file
ruff check app/main.py

Ruff Configuration (.pyproject.toml)

[tool.ruff]
target-version = "py312"
line-length = 88
select = [
    "E",   # pycodestyle errors
    "W",   # pycodestyle warnings
    "F",   # pyflakes
    "I",   # isort
    "B",   # flake8-bugbear
    "C4",  # flake8-comprehensions
    "UP",  # pyupgrade
]
ignore = [
    "E501",  # line too long, handled by black
    "B008",  # do not perform function calls in argument defaults
    "C901",  # too complex
]
exclude = [
    ".bzr",
    ".direnv",
    ".eggs",
    ".git",
    ".hg",
    ".mypy_cache",
    ".nox",
    ".pants.d",
    ".pytype",
    ".ruff_cache",
    ".svn",
    ".tox",
    ".venv",
    "__pypackages__",
    "_build",
    "buck-out",
    "build",
    "dist",
    "node_modules",
    "venv",
]

[tool.ruff.mccabe]
max-complexity = 10

[tool.ruff.isort]
known-first-party = ["app"]

4. Type Checking with mypy

# Install mypy
pip install mypy

# Check types
mypy app/

# Check specific file
mypy app/main.py

mypy Configuration (mypy.ini)

[mypy]
python_version = 3.12
warn_return_any = True
warn_unused_configs = True
disallow_untyped_defs = True
disallow_incomplete_defs = True
check_untyped_defs = True
disallow_untyped_decorators = True
no_implicit_optional = True
warn_redundant_casts = True
warn_unused_ignores = True
warn_no_return = True
warn_unreachable = True
strict_equality = True

[mypy-tests.*]
disallow_untyped_defs = False
disallow_incomplete_defs = False

[mypy-alembic.*]
ignore_errors = True

FastAPI Coding Standards

1. API Endpoint Structure

# Good: Clear, consistent endpoint structure
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List

from app.core.database import get_db
from app.core.security import get_current_user
from app.models.user import User
from app.schemas.application import ApplicationCreate, ApplicationResponse
from app.crud.application import create_application, get_user_applications

router = APIRouter(prefix="/api/v1/applications", tags=["applications"])

@router.post("/", response_model=ApplicationResponse, status_code=status.HTTP_201_CREATED)
async def create_job_application(
    application_data: ApplicationCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> ApplicationResponse:
    """
    Create a new job application with AI-generated cover letter.
    
    Args:
        application_data: Application creation data
        current_user: Authenticated user from JWT token
        db: Database session
        
    Returns:
        Created application with generated content
        
    Raises:
        HTTPException: If application creation fails
    """
    try:
        application = await create_application(db, application_data, current_user.id)
        return ApplicationResponse.from_orm(application)
    except Exception as e:
        logger.error(f"Failed to create application: {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to create application"
        )

@router.get("/", response_model=List[ApplicationResponse])
async def get_applications(
    skip: int = 0,
    limit: int = 100,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> List[ApplicationResponse]:
    """Get user's job applications with pagination."""
    
    applications = await get_user_applications(
        db, user_id=current_user.id, skip=skip, limit=limit
    )
    return [ApplicationResponse.from_orm(app) for app in applications]

2. Error Handling Standards

# Good: Consistent error handling
from app.core.exceptions import JobForgeException

class ApplicationNotFoundError(JobForgeException):
    """Raised when application is not found."""
    pass

class ApplicationAccessDeniedError(JobForgeException):
    """Raised when user doesn't have access to application."""
    pass

@router.get("/{application_id}", response_model=ApplicationResponse)
async def get_application(
    application_id: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> ApplicationResponse:
    """Get specific job application by ID."""
    
    try:
        application = await get_application_by_id(db, application_id, current_user.id)
        
        if not application:
            raise ApplicationNotFoundError(f"Application {application_id} not found")
            
        return ApplicationResponse.from_orm(application)
        
    except ApplicationNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Application not found"
        )
    except Exception as e:
        logger.error(f"Error retrieving application {application_id}: {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Internal server error"
        )

3. Dependency Injection

# Good: Proper dependency injection
from app.services.ai.claude_service import ClaudeService
from app.services.ai.openai_service import OpenAIService

async def get_claude_service() -> ClaudeService:
    """Dependency for Claude AI service."""
    return ClaudeService()

async def get_openai_service() -> OpenAIService:
    """Dependency for OpenAI service."""
    return OpenAIService()

@router.post("/{application_id}/generate-cover-letter")
async def generate_cover_letter(
    application_id: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    claude_service: ClaudeService = Depends(get_claude_service),
) -> dict:
    """Generate AI cover letter for application."""
    
    application = await get_application_by_id(db, application_id, current_user.id)
    if not application:
        raise HTTPException(status_code=404, detail="Application not found")
    
    cover_letter = await claude_service.generate_cover_letter(
        user_profile=current_user.profile,
        job_description=application.job_description
    )
    
    application.cover_letter = cover_letter
    await db.commit()
    
    return {"cover_letter": cover_letter}

Pydantic Model Standards

1. Schema Definitions

# Good: Clear schema definitions with validation
from pydantic import BaseModel, Field, EmailStr, validator
from typing import Optional, List
from datetime import datetime
from enum import Enum

class ApplicationStatus(str, Enum):
    """Application status enumeration."""
    DRAFT = "draft"
    APPLIED = "applied"
    INTERVIEW = "interview"
    REJECTED = "rejected"
    OFFER = "offer"

class ApplicationBase(BaseModel):
    """Base application schema."""
    company_name: str = Field(..., min_length=1, max_length=255, description="Company name")
    role_title: str = Field(..., min_length=1, max_length=255, description="Job role title")
    job_description: Optional[str] = Field(None, max_length=5000, description="Job description")
    status: ApplicationStatus = Field(ApplicationStatus.DRAFT, description="Application status")

class ApplicationCreate(ApplicationBase):
    """Schema for creating applications."""
    
    @validator('company_name')
    def validate_company_name(cls, v):
        if not v.strip():
            raise ValueError('Company name cannot be empty')
        return v.strip()
    
    @validator('role_title')
    def validate_role_title(cls, v):
        if not v.strip():
            raise ValueError('Role title cannot be empty')
        return v.strip()

class ApplicationUpdate(BaseModel):
    """Schema for updating applications."""
    company_name: Optional[str] = Field(None, min_length=1, max_length=255)
    role_title: Optional[str] = Field(None, min_length=1, max_length=255)
    job_description: Optional[str] = Field(None, max_length=5000)
    status: Optional[ApplicationStatus] = None

class ApplicationResponse(ApplicationBase):
    """Schema for application responses."""
    id: str = Field(..., description="Application ID")
    user_id: str = Field(..., description="User ID")
    cover_letter: Optional[str] = Field(None, description="Generated cover letter")
    created_at: datetime = Field(..., description="Creation timestamp")
    updated_at: datetime = Field(..., description="Last update timestamp")
    
    class Config:
        from_attributes = True  # For SQLAlchemy model conversion

2. Model Validation

# Good: Custom validation methods
from pydantic import BaseModel, validator, root_validator
import re

class UserCreate(BaseModel):
    """User creation schema with validation."""
    email: EmailStr
    password: str = Field(..., min_length=8, max_length=128)
    first_name: str = Field(..., min_length=1, max_length=50)
    last_name: str = Field(..., min_length=1, max_length=50)
    
    @validator('password')
    def validate_password_strength(cls, v):
        """Validate password strength."""
        if len(v) < 8:
            raise ValueError('Password must be at least 8 characters long')
        if not re.search(r'[A-Z]', v):
            raise ValueError('Password must contain at least one uppercase letter')
        if not re.search(r'[a-z]', v):
            raise ValueError('Password must contain at least one lowercase letter')
        if not re.search(r'\d', v):
            raise ValueError('Password must contain at least one digit')
        return v
    
    @validator('first_name', 'last_name')
    def validate_names(cls, v):
        """Validate name fields."""
        if not v.strip():
            raise ValueError('Name cannot be empty')
        if not re.match(r'^[a-zA-Z\s\'-]+$', v):
            raise ValueError('Name contains invalid characters')
        return v.strip().title()

Database Model Standards

1. SQLAlchemy Models

# Good: Well-structured SQLAlchemy models
from sqlalchemy import Column, String, Text, DateTime, ForeignKey, Enum
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
import uuid

from app.core.database import Base
from app.models.application import ApplicationStatus

class User(Base):
    """User model with proper relationships and constraints."""
    
    __tablename__ = "users"
    
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True)
    email = Column(String(255), unique=True, nullable=False, index=True)
    password_hash = Column(String(255), nullable=False)
    first_name = Column(String(100), nullable=False)
    last_name = Column(String(100), nullable=False)
    is_active = Column(Boolean, default=True, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
    
    # Relationships
    applications = relationship("Application", back_populates="user", cascade="all, delete-orphan")
    
    def __repr__(self) -> str:
        return f"<User(id={self.id}, email={self.email})>"

class Application(Base):
    """Application model with RLS and proper indexing."""
    
    __tablename__ = "applications"
    
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True)
    user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False, index=True)
    company_name = Column(String(255), nullable=False, index=True)
    role_title = Column(String(255), nullable=False)
    job_description = Column(Text)
    cover_letter = Column(Text)
    status = Column(Enum(ApplicationStatus), default=ApplicationStatus.DRAFT, nullable=False, index=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False, index=True)
    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
    
    # Relationships
    user = relationship("User", back_populates="applications")
    
    def __repr__(self) -> str:
        return f"<Application(id={self.id}, company={self.company_name}, status={self.status})>"

2. Database Operations (CRUD)

# Good: Async database operations with proper error handling
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update, delete
from sqlalchemy.orm import selectinload
from typing import Optional, List

from app.models.application import Application
from app.schemas.application import ApplicationCreate, ApplicationUpdate

async def create_application(
    db: AsyncSession, 
    application_data: ApplicationCreate, 
    user_id: str
) -> Application:
    """Create a new job application."""
    
    application = Application(
        user_id=user_id,
        **application_data.dict()
    )
    
    db.add(application)
    await db.commit()
    await db.refresh(application)
    
    return application

async def get_application_by_id(
    db: AsyncSession, 
    application_id: str, 
    user_id: str
) -> Optional[Application]:
    """Get application by ID with user validation."""
    
    query = select(Application).where(
        Application.id == application_id,
        Application.user_id == user_id
    )
    
    result = await db.execute(query)
    return result.scalar_one_or_none()

async def get_user_applications(
    db: AsyncSession,
    user_id: str,
    skip: int = 0,
    limit: int = 100,
    status_filter: Optional[ApplicationStatus] = None
) -> List[Application]:
    """Get user applications with filtering and pagination."""
    
    query = select(Application).where(Application.user_id == user_id)
    
    if status_filter:
        query = query.where(Application.status == status_filter)
    
    query = query.offset(skip).limit(limit).order_by(Application.created_at.desc())
    
    result = await db.execute(query)
    return list(result.scalars().all())

async def update_application(
    db: AsyncSession,
    application_id: str,
    application_data: ApplicationUpdate,
    user_id: str
) -> Optional[Application]:
    """Update application with user validation."""
    
    # Update only provided fields
    update_data = application_data.dict(exclude_unset=True)
    if not update_data:
        return None
    
    query = (
        update(Application)
        .where(Application.id == application_id, Application.user_id == user_id)
        .values(**update_data)
        .returning(Application)
    )
    
    result = await db.execute(query)
    await db.commit()
    
    return result.scalar_one_or_none()

Async Programming Standards

1. Async/Await Usage

# Good: Proper async/await usage
import asyncio
from typing import List, Optional

async def process_multiple_applications(
    applications: List[Application],
    ai_service: ClaudeService
) -> List[str]:
    """Process multiple applications concurrently."""
    
    async def process_single_application(app: Application) -> str:
        """Process a single application."""
        if not app.job_description:
            return ""
        
        return await ai_service.generate_cover_letter(
            user_profile=app.user.profile,
            job_description=app.job_description
        )
    
    # Process applications concurrently
    tasks = [process_single_application(app) for app in applications]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Handle exceptions
    cover_letters = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            logger.error(f"Failed to process application {applications[i].id}: {result}")
            cover_letters.append("")
        else:
            cover_letters.append(result)
    
    return cover_letters

2. Context Managers

# Good: Proper async context manager usage
from contextlib import asynccontextmanager
from typing import AsyncGenerator

@asynccontextmanager
async def get_ai_service_with_retry(
    max_retries: int = 3
) -> AsyncGenerator[ClaudeService, None]:
    """Context manager for AI service with retry logic."""
    
    service = ClaudeService()
    retries = 0
    
    try:
        while retries < max_retries:
            try:
                await service.test_connection()
                yield service
                break
            except Exception as e:
                retries += 1
                if retries >= max_retries:
                    raise e
                await asyncio.sleep(2 ** retries)  # Exponential backoff
    finally:
        await service.close()

# Usage
async def generate_with_retry(job_description: str) -> str:
    async with get_ai_service_with_retry() as ai_service:
        return await ai_service.generate_cover_letter(
            user_profile={}, 
            job_description=job_description
        )

Testing Standards

1. Test Structure

# Good: Well-structured tests
import pytest
from httpx import AsyncClient
from unittest.mock import AsyncMock, patch

class TestApplicationAPI:
    """Test suite for application API endpoints."""
    
    @pytest.mark.asyncio
    async def test_create_application_success(
        self, 
        async_client: AsyncClient, 
        test_user_token: str
    ):
        """Test successful application creation."""
        
        # Arrange
        application_data = {
            "company_name": "Test Corp",
            "role_title": "Software Developer",
            "job_description": "Python developer position",
            "status": "draft"
        }
        headers = {"Authorization": f"Bearer {test_user_token}"}
        
        # Act
        response = await async_client.post(
            "/api/v1/applications/",
            json=application_data,
            headers=headers
        )
        
        # Assert
        assert response.status_code == 201
        data = response.json()
        assert data["company_name"] == "Test Corp"
        assert data["role_title"] == "Software Developer"
        assert data["status"] == "draft"
        assert "id" in data
        assert "created_at" in data
    
    @pytest.mark.asyncio
    async def test_create_application_with_ai_generation(
        self,
        async_client: AsyncClient,
        test_user_token: str,
        mock_claude_service: AsyncMock
    ):
        """Test application creation with AI cover letter generation."""
        
        # Arrange
        mock_claude_service.generate_cover_letter.return_value = "Mock cover letter"
        
        application_data = {
            "company_name": "AI Corp",
            "role_title": "ML Engineer",
            "job_description": "Machine learning position with Python",
            "status": "draft"
        }
        headers = {"Authorization": f"Bearer {test_user_token}"}
        
        # Act
        with patch('app.services.ai.claude_service.ClaudeService', return_value=mock_claude_service):
            response = await async_client.post(
                "/api/v1/applications/",
                json=application_data,
                headers=headers
            )
        
        # Assert
        assert response.status_code == 201
        data = response.json()
        assert data["cover_letter"] == "Mock cover letter"
        mock_claude_service.generate_cover_letter.assert_called_once()

2. Test Fixtures

# Good: Reusable test fixtures
import pytest
from typing import AsyncGenerator
from httpx import AsyncClient
from fastapi.testclient import TestClient

@pytest.fixture
async def test_application(
    test_db: AsyncSession, 
    test_user: User
) -> Application:
    """Create a test application."""
    
    from app.crud.application import create_application
    from app.schemas.application import ApplicationCreate
    
    app_data = ApplicationCreate(
        company_name="Test Company",
        role_title="Test Role",
        job_description="Test job description",
        status="draft"
    )
    
    application = await create_application(test_db, app_data, test_user.id)
    await test_db.commit()
    
    return application

@pytest.fixture
def mock_ai_services():
    """Mock all AI services."""
    
    with patch('app.services.ai.claude_service.ClaudeService') as mock_claude, \
         patch('app.services.ai.openai_service.OpenAIService') as mock_openai:
        
        mock_claude.return_value.generate_cover_letter = AsyncMock(
            return_value="Mock cover letter"
        )
        mock_openai.return_value.create_embedding = AsyncMock(
            return_value=[0.1] * 1536
        )
        
        yield {
            'claude': mock_claude.return_value,
            'openai': mock_openai.return_value
        }

Documentation Standards

1. Docstring Format

# Good: Comprehensive docstrings
def calculate_job_match_score(
    user_skills: List[str],
    job_requirements: List[str],
    experience_years: int
) -> float:
    """
    Calculate job match score based on skills and experience.
    
    Args:
        user_skills: List of user's skills
        job_requirements: List of job requirements
        experience_years: Years of relevant experience
        
    Returns:
        Match score between 0.0 and 1.0
        
    Raises:
        ValueError: If experience_years is negative
        
    Example:
        >>> calculate_job_match_score(
        ...     ["Python", "FastAPI"], 
        ...     ["Python", "Django"], 
        ...     3
        ... )
        0.75
    """
    if experience_years < 0:
        raise ValueError("Experience years cannot be negative")
    
    # Implementation...
    return 0.75

2. API Documentation

# Good: Comprehensive API documentation
@router.post(
    "/",
    response_model=ApplicationResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Create job application",
    description="Create a new job application with optional AI-generated cover letter",
    responses={
        201: {"description": "Application created successfully"},
        400: {"description": "Invalid application data"},
        401: {"description": "Authentication required"},
        422: {"description": "Validation error"},
        500: {"description": "Internal server error"}
    }
)
async def create_job_application(
    application_data: ApplicationCreate = Body(
        ...,
        example={
            "company_name": "Google",
            "role_title": "Senior Python Developer",
            "job_description": "We are looking for an experienced Python developer...",
            "status": "draft"
        }
    ),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> ApplicationResponse:
    """Create a new job application."""
    # Implementation...

Performance Standards

1. Database Query Optimization

# Good: Optimized database queries
async def get_applications_with_stats(
    db: AsyncSession,
    user_id: str
) -> dict:
    """Get applications with statistics in a single query."""
    
    from sqlalchemy import func, case
    
    query = select(
        func.count(Application.id).label('total_applications'),
        func.count(case((Application.status == 'applied', 1))).label('applied_count'),
        func.count(case((Application.status == 'interview', 1))).label('interview_count'),
        func.count(case((Application.status == 'offer', 1))).label('offer_count'),
        func.avg(
            case((Application.created_at.isnot(None), 
                  func.extract('epoch', func.now() - Application.created_at)))
        ).label('avg_age_days')
    ).where(Application.user_id == user_id)
    
    result = await db.execute(query)
    row = result.first()
    
    return {
        'total_applications': row.total_applications or 0,
        'applied_count': row.applied_count or 0,
        'interview_count': row.interview_count or 0,
        'offer_count': row.offer_count or 0,
        'avg_age_days': round((row.avg_age_days or 0) / 86400, 1)  # Convert to days
    }

2. Caching Strategies

# Good: Implement caching for expensive operations
from functools import lru_cache
from typing import Dict, Any
import asyncio

@lru_cache(maxsize=128)
def get_job_keywords(job_description: str) -> List[str]:
    """Extract keywords from job description (cached)."""
    # Expensive NLP processing here
    return extract_keywords(job_description)

class CachedAIService:
    """AI service with caching."""
    
    def __init__(self):
        self._cache: Dict[str, Any] = {}
        self._cache_ttl = 3600  # 1 hour
    
    async def generate_cover_letter_cached(
        self, 
        user_profile: dict, 
        job_description: str
    ) -> str:
        """Generate cover letter with caching."""
        
        cache_key = f"{hash(str(user_profile))}_{hash(job_description)}"
        
        if cache_key in self._cache:
            cached_result, timestamp = self._cache[cache_key]
            if time.time() - timestamp < self._cache_ttl:
                return cached_result
        
        # Generate new cover letter
        result = await self.generate_cover_letter(user_profile, job_description)
        
        # Cache result
        self._cache[cache_key] = (result, time.time())
        
        return result

These coding standards ensure that Job Forge maintains high code quality, consistency, and performance across all components of the application.