Development Guide

git checkout -b feature/your-feature-name

# Use your IDE
code .  # VS Code
charm . # PyCharm

# Run tests for changed service
just test-service services/ai_service
 
# Run all tests
just test

# Auto-fix and format code
just lint
 
# Or run manually
uv run ruff check --fix
uv run ruff format

# Pre-commit hooks will run automatically
git add .
git commit -m "feat: add new diagnosis endpoint"

services/ai_service/src/
├── api/
│   └── v1/
│       ├── diagnosis.py      # Route handlers
│       └── image_quality.py
├── core/
│   ├── diagnostics/          # Business logic
│   │   ├── predictor.py
│   │   └── predictor_test.py # Tests co-located
│   └── image_quality/
├── schemas/
│   └── diagnosis.py          # Pydantic models
├── config.py                 # Configuration
└── main.py                   # FastAPI app

from pydantic import BaseModel
 
class MyFeatureRequest(BaseModel):
    input: str
 
class MyFeatureResponse(BaseModel):
    result: str

def process_my_feature(input: str) -> str:
    # Your logic here
    return f"Processed: {input}"

from fastapi import APIRouter
from schemas.my_feature import MyFeatureRequest, MyFeatureResponse
from core.my_feature import process_my_feature
 
router = APIRouter()
 
@router.post("/my-feature", response_model=MyFeatureResponse)
async def my_feature_endpoint(request: MyFeatureRequest):
    result = process_my_feature(request.input)
    return MyFeatureResponse(result=result)

from api.v1.my_feature import router as my_feature_router
 
app.include_router(my_feature_router, prefix="/api/v1", tags=["my-feature"])

# core/my_feature_test.py
from core.my_feature import process_my_feature
 
def test_process_my_feature():
    result = process_my_feature("test")
    assert result == "Processed: test"

# api/v1/my_feature_test.py
from fastapi.testclient import TestClient
from main import app
 
client = TestClient(app)
 
def test_my_feature_endpoint():
    response = client.post("/api/v1/my-feature", json={"input": "test"})
    assert response.status_code == 200
    assert response.json()["result"] == "Processed: test"

# Run all tests with coverage
just test
 
# Run specific test file
uv run pytest services/ai_service/src/core/my_feature_test.py
 
# Run with verbose output
uv run pytest -v
 
# Run and watch for changes
uv run pytest-watch

# packages/py_core/src/models/my_model.py
from sqlalchemy import Column, String, Integer
from db.base import Base
 
class MyModel(Base):
    __tablename__ = "my_table"
 
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)

# Auto-generate migration from model changes
just migrate-create "add_my_table"
 
# Review the generated migration
code packages/py_core/alembic/versions/xxx_add_my_table.py
 
# Apply migration
just migrate

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from models.my_model import MyModel
 
async def get_my_data(db: AsyncSession, id: int):
    result = await db.execute(
        select(MyModel).where(MyModel.id == id)
    )
    return result.scalar_one_or_none()

from typing import Optional
 
def my_function(name: str, age: int) -> Optional[str]:
    if age > 0:
        return f"{name} is {age}"
    return None

def my_function(param1: str, param2: int) -> bool:
    """
    Brief description of function.
 
    Args:
        param1: Description of param1
        param2: Description of param2
 
    Returns:
        Description of return value
 
    Raises:
        ValueError: When param2 is negative
    """
    if param2 < 0:
        raise ValueError("param2 must be positive")
    return True

def my_function(data: str):
    import pdb; pdb.set_trace()  # Debugger will stop here
    result = process_data(data)
    return result

import structlog
 
logger = structlog.get_logger(__name__)
 
def my_function():
    logger.info("processing_started", user_id=123)
    try:
        # Do work
        logger.info("processing_completed", result="success")
    except Exception as e:
        logger.error("processing_failed", error=str(e))
        raise

# View logs for all services
just logs
 
# View logs for specific service
just logs-service ai_service
 
# Follow logs in real-time
docker compose logs -f ai_service

import time
from fastapi import Request
 
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

from sqlalchemy import event
from sqlalchemy.engine import Engine
import time
 
@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time.time())
 
@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total = time.time() - conn.info['query_start_time'].pop(-1)
    print(f"Query took {total:.4f}s: {statement[:100]}")

# ✅ Good
async def get_user(db: AsyncSession, user_id: int):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()
 
# ❌ Bad
def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

from fastapi import Depends
from py_core.db.session import get_db
 
@router.get("/users/{user_id}")
async def get_user(
    user_id: int,
    db: AsyncSession = Depends(get_db)
):
    return await get_user_from_db(db, user_id)

from fastapi import HTTPException
 
@router.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await get_user_from_db(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

from pydantic import BaseModel, Field
 
class UserCreate(BaseModel):
    email: str = Field(..., regex=r"^[\w\.-]+@[\w\.-]+\.\w+$")
    age: int = Field(..., gt=0, lt=150)