SonarQube Code Quality Analysis
SonarQube วิเคราะห์ Code Quality ตรวจจับ Bugs, Code Smells, Vulnerabilities รองรับ 30+ ภาษา ใช้ Quality Gates กำหนดมาตรฐาน ใช้ร่วมกับ CI/CD Pipeline
Hexagonal Architecture แยก Domain ออกจาก External Systems ใช้ Ports and Adapters Pattern ทำให้ Code สะอาด ทดสอบง่าย SonarQube ช่วยตรวจสอบว่า Code ยังคง Clean
SonarQube Setup และ Configuration
# === SonarQube Setup ===
# 1. Docker Compose
# docker-compose.yml
# version: '3.8'
# services:
# sonarqube:
# image: sonarqube:community
# ports:
# - "9000:9000"
# environment:
# SONAR_JDBC_URL: jdbc:postgresql://db:5432/sonar
# SONAR_JDBC_USERNAME: sonar
# SONAR_JDBC_PASSWORD: sonar
# volumes:
# - sonarqube_data:/opt/sonarqube/data
# - sonarqube_extensions:/opt/sonarqube/extensions
# depends_on:
# - db
#
# db:
# image: postgres:16
# environment:
# POSTGRES_USER: sonar
# POSTGRES_PASSWORD: sonar
# POSTGRES_DB: sonar
# volumes:
# - postgresql_data:/var/lib/postgresql/data
# docker compose up -d
# เข้า http://localhost:9000 (admin/admin)
# 2. SonarScanner CLI
# npm install -g sonarqube-scanner
# หรือ
# brew install sonar-scanner
# 3. sonar-project.properties
cat > sonar-project.properties << 'EOF'
sonar.projectKey=my-hexagonal-app
sonar.projectName=My Hexagonal App
sonar.projectVersion=1.0
# Source
sonar.sources=src
sonar.tests=tests
sonar.sourceEncoding=UTF-8
# Language
sonar.language=py
sonar.python.version=3.11
# Coverage
sonar.python.coverage.reportPaths=coverage.xml
sonar.python.xunit.reportPath=test-results.xml
# Exclusions
sonar.exclusions=**/migrations/**,**/tests/**,**/__pycache__/**
sonar.test.exclusions=**/migrations/**
# Quality Gate
sonar.qualitygate.wait=true
EOF
# 4. รัน Analysis
# sonar-scanner \
# -Dsonar.host.url=http://localhost:9000 \
# -Dsonar.login=your-token
# 5. GitHub Actions Integration
# .github/workflows/sonar.yml
# name: SonarQube Analysis
# on: [push, pull_request]
# jobs:
# sonar:
# runs-on: ubuntu-latest
# steps:
# - uses: actions/checkout@v4
# with:
# fetch-depth: 0
# - uses: actions/setup-python@v5
# with:
# python-version: '3.11'
# - run: |
# pip install -r requirements.txt
# pytest --cov=src --cov-report=xml tests/
# - uses: SonarSource/sonarqube-scan-action@master
# env:
# SONAR_TOKEN: }
# SONAR_HOST_URL: }
echo "SonarQube configured"
echo " URL: http://localhost:9000"
echo " Scanner: sonar-scanner CLI"
echo " CI/CD: GitHub Actions integration"Hexagonal Architecture Implementation
# === Hexagonal Architecture ด้วย Python ===
# Project Structure:
# src/
# domain/ # Business Logic (ไม่พึ่ง External)
# models.py
# services.py
# ports.py # Interfaces
# adapters/ # External Implementations
# repositories/
# postgres_user_repo.py
# inmemory_user_repo.py
# notifications/
# email_service.py
# slack_service.py
# api/ # Input Adapters (REST, GraphQL)
# routes.py
# config/
# container.py # Dependency Injection
# === domain/models.py ===
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
from enum import Enum
class UserRole(Enum):
ADMIN = "admin"
MEMBER = "member"
VIEWER = "viewer"
@dataclass
class User:
id: Optional[str] = None
email: str = ""
name: str = ""
role: UserRole = UserRole.MEMBER
is_active: bool = True
created_at: datetime = field(default_factory=datetime.now)
def activate(self):
self.is_active = True
def deactivate(self):
self.is_active = False
def promote(self, role: UserRole):
if role == UserRole.ADMIN and self.role != UserRole.MEMBER:
raise ValueError("Only members can be promoted to admin")
self.role = role
# === domain/ports.py (Interfaces) ===
from abc import ABC, abstractmethod
from typing import List, Optional
class UserRepository(ABC):
"""Port สำหรับ User Data Access"""
@abstractmethod
def save(self, user: User) -> User:
pass
@abstractmethod
def find_by_id(self, user_id: str) -> Optional[User]:
pass
@abstractmethod
def find_by_email(self, email: str) -> Optional[User]:
pass
@abstractmethod
def find_all(self, limit: int = 100, offset: int = 0) -> List[User]:
pass
@abstractmethod
def delete(self, user_id: str) -> bool:
pass
class NotificationService(ABC):
"""Port สำหรับ Notifications"""
@abstractmethod
def send(self, to: str, subject: str, body: str) -> bool:
pass
class EventPublisher(ABC):
"""Port สำหรับ Domain Events"""
@abstractmethod
def publish(self, event_type: str, data: dict) -> None:
pass
# === domain/services.py (Business Logic) ===
class UserService:
"""Domain Service — Business Logic เท่านั้น"""
def __init__(self, repo: UserRepository, notifier: NotificationService,
publisher: EventPublisher):
self.repo = repo
self.notifier = notifier
self.publisher = publisher
def register_user(self, email: str, name: str) -> User:
existing = self.repo.find_by_email(email)
if existing:
raise ValueError(f"Email already registered: {email}")
user = User(email=email, name=name)
saved = self.repo.save(user)
self.notifier.send(email, "Welcome!", f"สวัสดี {name}")
self.publisher.publish("user.registered", {"user_id": saved.id})
return saved
def get_user(self, user_id: str) -> User:
user = self.repo.find_by_id(user_id)
if not user:
raise ValueError(f"User not found: {user_id}")
return user
def deactivate_user(self, user_id: str) -> User:
user = self.get_user(user_id)
user.deactivate()
self.repo.save(user)
self.publisher.publish("user.deactivated", {"user_id": user_id})
return user
print("Hexagonal Architecture:")
print(" Domain: Models, Services, Ports (no external deps)")
print(" Adapters: Repositories, Notifications, API")
print(" Ports: UserRepository, NotificationService, EventPublisher")Adapters และ Tests
# === Adapters (Implementations) ===
# adapters/repositories/inmemory_user_repo.py
import uuid
from typing import List, Optional
class InMemoryUserRepository:
"""In-memory Implementation สำหรับ Testing"""
def __init__(self):
self.users = {}
def save(self, user):
if not user.id:
user.id = str(uuid.uuid4())
self.users[user.id] = user
return user
def find_by_id(self, user_id):
return self.users.get(user_id)
def find_by_email(self, email):
for user in self.users.values():
if user.email == email:
return user
return None
def find_all(self, limit=100, offset=0):
all_users = list(self.users.values())
return all_users[offset:offset + limit]
def delete(self, user_id):
if user_id in self.users:
del self.users[user_id]
return True
return False
# adapters/notifications/fake_notifier.py
class FakeNotificationService:
"""Fake Notification สำหรับ Testing"""
def __init__(self):
self.sent = []
def send(self, to, subject, body):
self.sent.append({"to": to, "subject": subject, "body": body})
return True
# adapters/events/fake_publisher.py
class FakeEventPublisher:
"""Fake Event Publisher สำหรับ Testing"""
def __init__(self):
self.events = []
def publish(self, event_type, data):
self.events.append({"type": event_type, "data": data})
# === Tests ===
# tests/test_user_service.py
def test_register_user():
"""ทดสอบ Register User"""
repo = InMemoryUserRepository()
notifier = FakeNotificationService()
publisher = FakeEventPublisher()
service = UserService(repo, notifier, publisher)
user = service.register_user("test@example.com", "Test User")
assert user.id is not None
assert user.email == "test@example.com"
assert user.name == "Test User"
assert user.is_active is True
assert len(notifier.sent) == 1
assert len(publisher.events) == 1
assert publisher.events[0]["type"] == "user.registered"
print(" PASS: test_register_user")
def test_register_duplicate_email():
"""ทดสอบ Register Email ซ้ำ"""
repo = InMemoryUserRepository()
notifier = FakeNotificationService()
publisher = FakeEventPublisher()
service = UserService(repo, notifier, publisher)
service.register_user("test@example.com", "User 1")
try:
service.register_user("test@example.com", "User 2")
assert False, "Should raise ValueError"
except ValueError:
pass
print(" PASS: test_register_duplicate_email")
def test_deactivate_user():
"""ทดสอบ Deactivate User"""
repo = InMemoryUserRepository()
notifier = FakeNotificationService()
publisher = FakeEventPublisher()
service = UserService(repo, notifier, publisher)
user = service.register_user("test@example.com", "Test")
deactivated = service.deactivate_user(user.id)
assert deactivated.is_active is False
print(" PASS: test_deactivate_user")
# รัน Tests
print("\nRunning Tests:")
test_register_user()
test_register_duplicate_email()
test_deactivate_user()
print("\nAll tests passed!")Best Practices
- Quality Gates: ตั้ง Quality Gate ให้ Coverage > 80%, ไม่มี Critical Bugs
- Domain Isolation: Domain Layer ต้องไม่ Import อะไรจาก Adapters
- Port-first: ออกแบบ Port (Interface) ก่อน แล้วค่อยสร้าง Adapter
- Test with Fakes: ใช้ In-memory Adapters สำหรับ Unit Tests เร็วและ Isolated
- CI/CD Gate: รัน SonarQube ใน CI/CD Block Deploy ถ้าไม่ผ่าน Quality Gate
- Dependency Rule: Dependencies ชี้เข้าใน ไม่ออกนอก Domain ไม่รู้จัก Infrastructure
SonarQube คืออะไร
Open-source Platform วิเคราะห์ Code Quality ตรวจจับ Bugs Code Smells Vulnerabilities Duplications รองรับ 30+ ภาษา Quality Gates มาตรฐาน CI/CD Community Edition ฟรี
Hexagonal Architecture คืออะไร
Ports and Adapters แยก Business Logic ออกจาก External Systems ใช้ Ports Interfaces เป็นตัวกลาง Adapters เชื่อม External ทดสอบง่าย เปลี่ยน Technology ไม่กระทบ Domain
Quality Gate คืออะไร
เกณฑ์ว่า Code ผ่านมาตรฐานหรือไม่ Coverage 80%+ ไม่มี Critical Bugs Code Smells ไม่เกิน 5 Duplications ต่ำกว่า 3% ไม่ผ่าน Block Deploy CI/CD
Ports and Adapters Pattern ทำงานอย่างไร
Port เป็น Interface ที่ Domain กำหนด เช่น UserRepository Adapter เป็น Implementation เชื่อม External เช่น PostgresUserRepository Domain ใช้แค่ Port เปลี่ยน Adapter ได้ง่าย
สรุป
SonarQube ร่วมกับ Hexagonal Architecture ให้ Code ที่มีคุณภาพและ Architecture ที่สะอาด Quality Gates กำหนดมาตรฐาน Domain Isolation แยก Business Logic Ports and Adapters เชื่อม External Systems Test with Fakes สำหรับ Unit Tests CI/CD Gate Block Deploy ถ้าไม่ผ่าน
