guardia-docs/03_개발자지침서.md
DESKTOP-TKLFCPRython 938b25f286 feat(itsm): G-1~G-12 확장 기능 + 하네스/봇/설치스크립트 구현
G-1: 메신저 Webhook Relay + _send_to_room 실제 httpx 호출 구현
G-2: POST /api/tasks/bulk SR 대량작업 엔드포인트 (최대 100건)
G-3: 라이선스 만료 알림 스케줄러 (매일 09:00 KST)
G-4: 체험판 upgrade_banner 필드 + license.py 배너 로직
G-5: core/auto_rca.py + incidents/problem auto-rca 엔드포인트
G-6: core/deploy_impact.py + vibe impact-analysis 엔드포인트
G-7: core/ticket_classifier.py + SR 생성 시 AI 분류 + ai-suggestion API
G-8: VulnPatchRecord 모델 + vuln_scan 패치추적 4개 엔드포인트
G-9: core/jira_sync.py + gateway Jira/Confluence 연동 엔드포인트
G-10: core/push_notify.py + routers/push.py + PushSubscription 모델
G-11: approvals 다중승인 (위임/서명/기한초과/마감연장)
G-12: alembic.ini + migrations/ + cicd/migrate_to_postgres.sh

하네스: guardia-orchestrator 확장기능 Phase 반영
봇명령어: /sr /status /license /bulk 슬래시 명령어 추가
설치스크립트: setup/ (Ubuntu, CentOS, RHEL, Windows) --test 옵션 포함

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-29 18:18:52 +09:00

952 lines
26 KiB
Markdown

# GUARDiA ITSM + Messenger — 개발자 지침서
**문서 버전**: 1.0
**작성일**: 2026-05-25
**대상**: GUARDiA 플랫폼 개발자
---
## 목차
1. [개발 환경 설정](#1-개발-환경-설정)
2. [프로젝트 구조](#2-프로젝트-구조)
3. [코딩 컨벤션](#3-코딩-컨벤션)
4. [API 개발 가이드](#4-api-개발-가이드)
5. [데이터베이스 가이드](#5-데이터베이스-가이드)
6. [보안 필수 규칙](#6-보안-필수-규칙)
7. [테스트 작성 가이드](#7-테스트-작성-가이드)
8. [Messenger Bot 개발 가이드](#8-messenger-bot-개발-가이드)
9. [스케줄러 확장 가이드](#9-스케줄러-확장-가이드)
10. [배포 및 CI/CD](#10-배포-및-cicd)
11. [자주 하는 실수 및 주의사항](#11-자주-하는-실수-및-주의사항)
---
## 1. 개발 환경 설정
### 1.1 필수 도구
```
Python 3.11+
Git
VS Code (권장) 또는 PyCharm
SQLite Browser (DB 확인용)
httpie 또는 Postman (API 테스트)
```
### 1.2 로컬 개발 환경 구성
```bash
# 1. 레포지토리 클론
git clone <repo-url>
cd GUARDiA/itsm
# 2. 가상환경 생성 (Python 3.11 사용)
python -m venv .venv
# Windows
.venv\Scripts\activate
# Linux/Mac
source .venv/bin/activate
# 3. 의존성 설치
pip install -r requirements.txt
pip install -r requirements-dev.txt # 개발/테스트용
# 4. 환경 변수 설정
cp .env.example .env
# .env 파일 편집 (SECRET_KEY, DB_PATH 등)
# 5. 앱 실행
uvicorn main:app --reload --port 8000
# 6. API 문서 확인
# http://localhost:8000/docs
```
### 1.3 .env 파일 설정
```ini
# .env.example
SECRET_KEY=your-256-bit-secret-key-here-change-in-production
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=480
DB_URL=sqlite+aiosqlite:///./guardia_itsm.db
# 암호화 키 (AES-256-GCM, 32바이트 hex)
ENCRYPTION_KEY=0000000000000000000000000000000000000000000000000000000000000000
# 라이선스 마스터 키 (64자리 hex, 32바이트)
# 생성: python -c "import secrets; print(secrets.token_hex(32))"
# 운영 환경에서는 반드시 실제 값으로 교체 — 분실 시 기발급 라이선스 전부 무효화됨
GUARDIA_LICENSE_KEY=0000000000000000000000000000000000000000000000000000000000000000
# Messenger 연동
MESSENGER_BASE_URL=http://localhost:8002
MESSENGER_BOT_TOKEN=dev-token
# SSH 실행 타임아웃 (초)
SSH_TIMEOUT=30
# 파일 업로드 경로
UPLOAD_ROOT=./uploads
MAX_FILE_SIZE_MB=10
```
> **주의**: `.env` 파일은 절대 Git에 커밋하지 마세요. `.gitignore`에 포함되어 있습니다.
### 1.4 VS Code 권장 설정
`.vscode/settings.json`:
```json
{
"python.defaultInterpreterPath": ".venv/Scripts/python.exe",
"editor.formatOnSave": true,
"python.formatting.provider": "black",
"python.linting.enabled": true,
"python.linting.pylintEnabled": false,
"python.linting.flake8Enabled": true,
"python.linting.flake8Args": ["--max-line-length=120"]
}
```
---
## 2. 프로젝트 구조
```
C:\GUARDiA\itsm\
├── main.py # FastAPI 앱 진입점
├── database.py # DB 엔진, SessionLocal, get_db
├── models.py # SQLAlchemy ORM 모델 (전체)
├── schemas.py # Pydantic 스키마 (전체)
├── requirements.txt # 운영 의존성
├── requirements-dev.txt # 개발/테스트 의존성
├── .env # 환경 변수 (git 제외)
├── core\
│ ├── seed.py # 초기 데이터 시드
│ ├── scheduler.py # APScheduler 백그라운드 작업
│ └── security.py # JWT, 비밀번호 해싱
├── routers\
│ ├── auth.py # 인증/권한
│ ├── tasks.py # SR 처리
│ ├── approvals.py # 결재
│ ├── ssl_manager.py # SSL 인증서 관리
│ ├── pm.py # 정기점검
│ ├── incidents.py # 장애 관리
│ ├── oncall.py # 온콜/당직
│ ├── batch.py # 배치 작업
│ └── ... # 기타 라우터
├── utils\
│ ├── crypto.py # AES-256-GCM 암/복호화
│ └── ssh_runner.py # SSH 명령 실행
├── static\ # 프론트엔드 (HTML/CSS/JS)
├── uploads\ # 업로드 파일 저장소
│ ├── sr_files\
│ └── workspaces\
├── scripts\sm\ssl\
│ └── ssl_expiry_check.sh # 배포용 SSL 점검 스크립트
└── tests\
├── conftest.py
├── unit\
└── integration\
C:\GUARDiA\messenger\
├── main.py # Messenger 서버
├── models\ # DB 모델
├── core\ # 봇 로직
└── routers\ # API 라우터
```
---
## 3. 코딩 컨벤션
### 3.1 Python 스타일
```python
# 파일 맨 위: 임포트 순서 (isort 준수)
# 1) 표준 라이브러리
import os
import json
from datetime import datetime, timedelta
# 2) 서드파티
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
# 3) 로컬 모듈
from database import get_db
from models import SRRequest
from schemas import SRCreate, SROut
# 줄 최대 길이: 120자
# 들여쓰기: 4 space (탭 금지)
# 문자열: 큰따옴표(") 사용
```
### 3.2 라우터 파일 구조 패턴
새 라우터를 만들 때 반드시 이 구조를 따르세요:
```python
# routers/example.py
from datetime import datetime
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from database import get_db
from models import ExampleModel
from schemas import ExampleCreate, ExampleOut, ExampleUpdate
from core.security import get_current_user
from models import User, Role
router = APIRouter(prefix="/example", tags=["example"])
def _require_role(*roles: Role):
"""역할 기반 접근 제어 의존성"""
async def checker(current_user: User = Depends(get_current_user)):
if current_user.role not in roles:
raise HTTPException(status_code=403, detail="권한이 없습니다.")
return current_user
return checker
@router.get("/", response_model=list[ExampleOut])
async def list_examples(
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
result = await db.execute(select(ExampleModel).order_by(ExampleModel.created_at.desc()))
return result.scalars().all()
@router.post("/", response_model=ExampleOut, status_code=status.HTTP_201_CREATED)
async def create_example(
body: ExampleCreate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(_require_role(Role.ADMIN, Role.PM)),
):
obj = ExampleModel(**body.model_dump(), created_by=current_user.id)
db.add(obj)
await db.commit()
await db.refresh(obj)
return obj
```
### 3.3 모델 정의 패턴
```python
# models.py 에 추가할 때
class ExampleModel(Base):
__tablename__ = "tb_example"
id = Column(Integer, primary_key=True, autoincrement=True)
title = Column(String(200), nullable=False)
description = Column(Text, nullable=True)
is_active = Column(Boolean, default=True, nullable=False)
created_by = Column(Integer, ForeignKey("tb_user.id"), nullable=True)
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
```
### 3.4 스키마 정의 패턴
```python
# schemas.py 에 추가할 때
class ExampleCreate(BaseModel):
title: str = Field(..., min_length=1, max_length=200)
description: Optional[str] = None
class ExampleUpdate(BaseModel):
title: Optional[str] = Field(None, min_length=1, max_length=200)
description: Optional[str] = None
class ExampleOut(BaseModel):
id: int
title: str
description: Optional[str]
is_active: bool
created_at: datetime
model_config = ConfigDict(from_attributes=True)
```
### 3.5 명명 규칙
| 대상 | 규칙 | 예시 |
|------|------|------|
| 파일명 | snake_case | `ssl_manager.py` |
| 클래스명 | PascalCase | `SslDomain`, `PmSchedule` |
| 함수/변수 | snake_case | `get_ssl_domains()`, `days_left` |
| DB 테이블 | `tb_` 접두사 | `tb_ssl_domain` |
| 상수 | UPPER_SNAKE | `MAX_FILE_SIZE`, `SSL_WARN_DAYS` |
| 비공개 함수 | `_` 접두사 | `_notify_incident()` |
| API 경로 | kebab-case | `/ssl-manager/domains` |
---
## 4. API 개발 가이드
### 4.1 응답 형식 규칙
```python
# 성공 응답: HTTP 상태 코드로 구분
# 200 OK — 조회, 수정
# 201 Created — 생성
# 202 Accepted — 비동기 실행 (배치, 백그라운드)
# 204 No Content — 삭제
# 에러 응답: 반드시 이 형식 사용
raise HTTPException(
status_code=400,
detail="사용자 친화적인 한국어 메시지"
)
# 절대 금지: 스택트레이스 노출
# raise Exception(traceback.format_exc()) # ❌
```
### 4.2 페이징 구현 패턴
```python
from fastapi import Query
@router.get("/", response_model=dict)
async def list_items(
page: int = Query(1, ge=1),
size: int = Query(20, ge=1, le=100),
db: AsyncSession = Depends(get_db),
):
offset = (page - 1) * size
count_q = select(func.count()).select_from(Item)
total = (await db.execute(count_q)).scalar()
items_q = select(Item).offset(offset).limit(size).order_by(Item.created_at.desc())
items = (await db.execute(items_q)).scalars().all()
return {
"items": items,
"total": total,
"page": page,
"size": size,
"pages": (total + size - 1) // size,
}
```
### 4.3 비동기 DB 쿼리 패턴
```python
# ✅ 올바른 패턴: asyncio 방식
async def get_item(item_id: int, db: AsyncSession) -> Item:
result = await db.execute(select(Item).where(Item.id == item_id))
item = result.scalar_one_or_none()
if item is None:
raise HTTPException(status_code=404, detail="항목을 찾을 수 없습니다.")
return item
# ✅ 여러 조건 필터
query = select(Item).where(
Item.is_active == True,
Item.category == category,
).order_by(Item.created_at.desc())
# ✅ JOIN
query = select(Item).join(User, Item.created_by == User.id)
# ❌ 잘못된 패턴: 동기 방식 사용 금지
# db.query(Item).filter(Item.id == item_id).first() # ❌
```
### 4.4 백그라운드 작업 패턴
```python
from fastapi import BackgroundTasks
@router.post("/{id}/run", status_code=202)
async def run_job(
id: int,
background_tasks: BackgroundTasks,
db: AsyncSession = Depends(get_db),
):
job = await _get_job(id, db)
background_tasks.add_task(_execute_job, job.id)
return {"message": "실행이 요청되었습니다.", "job_id": id}
async def _execute_job(job_id: int):
# 중요: 새 DB 세션 생성 (request-scope 세션 재사용 금지!)
from database import SessionLocal
async with SessionLocal() as db:
job = await _get_job(job_id, db)
# ... 실행 로직
await db.commit()
```
### 4.5 main.py 라우터 등록
새 라우터를 추가할 때:
```python
# main.py
from routers import (
...
new_module, # ← 여기에 추가
)
# lifespan 함수 안에 필요한 초기화 추가
# 라우터 등록 (알파벳 순 권장)
app.include_router(new_module.router)
```
---
## 5. 데이터베이스 가이드
### 5.1 마이그레이션 정책
현재 GUARDiA ITSM은 **Alembic 없이** `init_db()`로 테이블을 자동 생성합니다.
```python
# database.py
async def init_db():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
```
> **주의**: 운영 DB에서는 테이블 삭제/재생성이 일어나지 않습니다. 컬럼 추가/변경 시 수동 ALTER TABLE이 필요합니다.
**신규 컬럼 추가 절차**:
1. `models.py`에 컬럼 추가
2. 운영 DB에 수동 ALTER: `ALTER TABLE tb_xxx ADD COLUMN new_col TEXT;`
3. 기본값 업데이트: `UPDATE tb_xxx SET new_col = 'default' WHERE new_col IS NULL;`
### 5.2 관계 설정 패턴
```python
# 1:N 관계 (부모→자식)
class SRRequest(Base):
__tablename__ = "tb_sr_request"
# ...
work_logs = relationship("WorkLog", back_populates="sr", lazy="select")
class WorkLog(Base):
__tablename__ = "tb_work_log"
sr_id = Column(Integer, ForeignKey("tb_sr_request.id"), nullable=False)
sr = relationship("SRRequest", back_populates="work_logs")
# 쿼리 시 eagerly load (N+1 문제 방지)
from sqlalchemy.orm import selectinload
query = select(SRRequest).options(selectinload(SRRequest.work_logs))
```
### 5.3 enum 정의 패턴
```python
import enum
class SRStatus(str, enum.Enum):
OPEN = "OPEN"
IN_PROGRESS = "IN_PROGRESS"
RESOLVED = "RESOLVED"
CLOSED = "CLOSED"
# 모델에서 사용
class SRRequest(Base):
status = Column(Enum(SRStatus), default=SRStatus.OPEN, nullable=False)
```
### 5.4 트랜잭션 처리
```python
# 여러 테이블 수정 시 트랜잭션 보장
async def complex_operation(db: AsyncSession):
try:
obj1 = Model1(...)
obj2 = Model2(...)
db.add(obj1)
db.add(obj2)
await db.commit() # 모두 성공 시 커밋
except Exception:
await db.rollback() # 하나라도 실패 시 롤백
raise
```
---
## 6. 보안 필수 규칙
> **이 규칙들은 절대 예외가 없습니다. 위반 시 코드 리뷰에서 반려됩니다.**
### 6.1 서버 자격증명 보호
```python
# ✅ 서버 조회 응답에서 민감 필드 제외
class ServerOut(BaseModel):
id: int
hostname: str
server_role: str
# ip_addr, ssh_user, os_pw_enc 절대 포함 금지!
# ❌ 절대 하지 말 것
class ServerOut(BaseModel):
ip_addr: str # ❌ IP 노출 금지
ssh_user: str # ❌ SSH 계정 노출 금지
os_pw_enc: str # ❌ 암호화된 비밀번호도 노출 금지
```
### 6.2 SSH 명령 안전성 검증
```python
_DANGEROUS_PATTERNS = [
"rm -rf /", "rm -rf /*", "mkfs", "dd if=",
"shutdown", "reboot", "halt", "poweroff",
":(){:|:&};:", ">()", "chmod 777 /",
"chown -R root /", "> /dev/sda",
]
def _validate_command(cmd: str) -> None:
for pattern in _DANGEROUS_PATTERNS:
if pattern in cmd:
raise HTTPException(
status_code=400,
detail=f"안전하지 않은 명령어 패턴이 포함되어 있습니다: {pattern}"
)
```
### 6.3 파일 경로 순회 방지
```python
from pathlib import Path
UPLOAD_ROOT = Path(settings.UPLOAD_ROOT).resolve()
def _safe_path(filename: str, subdir: str) -> Path:
target = (UPLOAD_ROOT / subdir / filename).resolve()
if not str(target).startswith(str(UPLOAD_ROOT)):
raise HTTPException(status_code=400, detail="잘못된 파일 경로입니다.")
return target
```
### 6.4 AES-256-GCM 암호화 사용
```python
# utils/crypto.py
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os, base64
def encrypt(plaintext: str, key_hex: str) -> str:
key = bytes.fromhex(key_hex)
nonce = os.urandom(12)
aesgcm = AESGCM(key)
ct = aesgcm.encrypt(nonce, plaintext.encode(), None)
return base64.b64encode(nonce + ct).decode()
def decrypt(ciphertext: str, key_hex: str) -> str:
key = bytes.fromhex(key_hex)
data = base64.b64decode(ciphertext)
nonce, ct = data[:12], data[12:]
aesgcm = AESGCM(key)
return aesgcm.decrypt(nonce, ct, None).decode()
```
### 6.5 Messenger 발송 메시지 보안
```python
# ✅ 올바른 알림 메시지 (서버 정보 제외)
message = f"[장애] {incident.title}\n등급: {incident.priority}\nID: {incident.incident_id}"
# ❌ 절대 금지 (서버 정보 포함)
message = f"서버 {server.ip_addr}에서 장애 발생" # ❌ IP 노출
message = f"SSH 계정: {server.ssh_user}" # ❌ 계정 노출
```
### 6.6 JWT 토큰 처리
```python
# 토큰 생성 시 만료 시간 필수
def create_access_token(data: dict) -> str:
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
return jwt.encode({**data, "exp": expire}, settings.SECRET_KEY, algorithm="HS256")
# 의존성에서 자동 검증
async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)):
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
except JWTError:
raise HTTPException(status_code=401, detail="유효하지 않은 토큰입니다.")
```
### 6.7 root SSH 접속 금지
```python
# 서버 등록/수정 시 검증
if body.ssh_user == "root":
raise HTTPException(status_code=400, detail="root SSH 직접 접속은 허용되지 않습니다.")
```
### 6.8 라이선스 제한 강제
에디션별 자원 한도(기관·사용자·서버 수)를 초과하는 생성 API에는 반드시 Dependency를 추가해야 한다.
```python
from middleware.license_guard import (
check_institution_limit,
check_server_limit,
check_user_limit,
require_feature,
)
# 기관 등록 엔드포인트 예시
@router.post("/", response_model=InstitutionOut, status_code=201)
async def create_institution(
payload: InstitutionCreate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
_: None = Depends(check_institution_limit), # 에디션 한도 초과 시 HTTP 403
):
...
# 서버 등록 엔드포인트 예시
@router.post("/servers", response_model=ServerOut, status_code=201)
async def create_server(
payload: dict,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
_: None = Depends(check_server_limit), # 에디션 한도 초과 시 HTTP 403
):
...
# 특정 기능이 에디션에 포함되는지 체크 (STANDARD 이상 필요한 LDAP 예시)
@router.post("/ldap/sync")
async def ldap_sync(
_: None = Depends(require_feature("LDAP")),
):
...
```
에디션별 기본 한도:
| 에디션 | 기관 | 사용자 | 서버 |
|--------|------|-------|------|
| COMMUNITY | 1 | 10 | 20 |
| STANDARD | 50 | 200 | 500 |
| ENTERPRISE | 무제한 | 무제한 | 무제한 |
---
## 7. 테스트 작성 가이드
### 7.1 단위 테스트 기본 구조
```python
import pytest
import pytest_asyncio
class TestSslManager:
@pytest.mark.asyncio
async def test_register_domain_success(self, client, auth_headers):
# Arrange
payload = {"domain": "example.com", "port": 443, "server_id": 1}
# Act
r = await client.post("/ssl-manager/domains", json=payload, headers=auth_headers)
# Assert
assert r.status_code == 201
data = r.json()
assert data["domain"] == "example.com"
assert "id" in data
@pytest.mark.asyncio
async def test_register_domain_duplicate(self, client, auth_headers):
payload = {"domain": "example.com", "port": 443, "server_id": 1}
await client.post("/ssl-manager/domains", json=payload, headers=auth_headers)
r = await client.post("/ssl-manager/domains", json=payload, headers=auth_headers)
assert r.status_code == 409
```
### 7.2 mock 사용 패턴
```python
from unittest.mock import AsyncMock, patch
@pytest.mark.asyncio
async def test_p1_incident_triggers_notification(client, auth_headers):
with patch("routers.incidents._notify_incident", new_callable=AsyncMock) as mock_notify:
r = await client.post("/incidents/", json={
"title": "DB 서버 다운",
"priority": "P1",
"description": "운영 DB 응답 없음",
}, headers=auth_headers)
assert r.status_code == 201
mock_notify.assert_called_once()
call_args = mock_notify.call_args
assert call_args[0][0].priority == "P1" # 첫 번째 인자 확인
```
### 7.3 시간 고정 테스트
```python
from freezegun import freeze_time
@pytest.mark.asyncio
@freeze_time("2026-01-15 00:00:00")
async def test_pm_schedule_monthly_next_date(client, auth_headers):
r = await client.post("/pm/schedules/", json={
"template_id": 1,
"frequency": "MONTHLY",
"day_of_month": 15,
"server_id": 1,
}, headers=auth_headers)
assert r.status_code == 201
# 1월 15일 기준 → 다음 실행은 2월 15일
assert "2026-02-15" in r.json()["next_scheduled"]
```
---
## 8. Messenger Bot 개발 가이드
### 8.1 GUARDiA_ITSM 봇 명령 추가
명령어 추가는 `C:\GUARDiA\messenger\core\itsm_bot.py` 에서 수행합니다.
```python
# 새 명령어 핸들러 추가 패턴
COMMAND_HANDLERS = {
"/itsm sr list": handle_sr_list,
"/itsm sr create": handle_sr_create,
"/itsm help": handle_help,
"/itsm incident p1": handle_p1_alert, # 새 명령어 추가
}
async def handle_p1_alert(args: list[str], user_id: int) -> str:
"""P1 장애 목록 조회"""
async with httpx.AsyncClient() as client:
r = await client.get(
f"{ITSM_BASE_URL}/incidents/?priority=P1&status=OPEN",
headers={"Authorization": f"Bearer {ITSM_BOT_TOKEN}"},
)
if r.status_code != 200:
return "장애 목록 조회 실패"
incidents = r.json()["items"]
if not incidents:
return "현재 P1 장애 없음"
lines = [f"[{i['incident_id']}] {i['title']}" for i in incidents[:5]]
return "P1 장애 목록:\n" + "\n".join(lines)
```
### 8.2 ITSM → Messenger 알림 발송
```python
# routers/incidents.py 내 알림 함수
async def _notify_incident(incident: IncidentModel) -> None:
"""장애 발생 시 Messenger 채널 알림"""
try:
channel_id = settings.INCIDENT_CHANNEL_ID
# 보안: 서버 IP/PW 미포함
msg = (
f"🚨 **{incident.priority} 장애 발생**\n"
f"ID: {incident.incident_id}\n"
f"제목: {incident.title}\n"
f"등록: {incident.created_at.strftime('%Y-%m-%d %H:%M')}"
)
async with httpx.AsyncClient(timeout=5) as client:
await client.post(
f"{settings.MESSENGER_BASE_URL}/api/messages",
json={"channel_id": channel_id, "content": msg},
headers={"Authorization": f"Bearer {settings.MESSENGER_BOT_TOKEN}"},
)
except Exception:
# 알림 실패가 비즈니스 로직을 방해하면 안 됨
pass
```
---
## 9. 스케줄러 확장 가이드
### 9.1 새 스케줄 작업 추가
`core/scheduler.py`에 새 작업을 추가합니다:
```python
from apscheduler.triggers.cron import CronTrigger
async def _new_scheduled_task() -> None:
"""매주 월요일 오전 9시 실행 예시"""
from database import SessionLocal
async with SessionLocal() as db:
# 작업 수행
pass
def start_scheduler():
# 기존 작업들 ...
_scheduler.add_job(
_new_scheduled_task,
CronTrigger(day_of_week="mon", hour=9, minute=0),
id="new_task",
replace_existing=True,
)
_scheduler.start()
```
### 9.2 스케줄러 테스트
스케줄러는 직접 함수 호출로 테스트합니다:
```python
@pytest.mark.asyncio
async def test_ssl_expiry_scan(db_session):
from core.scheduler import _scan_ssl_expiry
# 테스트 데이터 준비
# ...
await _scan_ssl_expiry() # 직접 호출
# DB 상태 확인
# ...
```
---
## 10. 배포 및 CI/CD
### 10.1 Jenkins 파이프라인 단계
```groovy
pipeline {
stages {
stage('Test') {
steps {
sh 'pytest tests/unit/ --cov=. --cov-report=xml'
}
}
stage('Quality Gate') {
steps {
// SonarQube 분석
sh 'sonar-scanner -Dsonar.projectKey=guardia-itsm'
}
}
stage('Deploy') {
when { branch 'main' }
steps {
sh './scripts/deploy.sh production'
}
}
}
}
```
### 10.2 운영 서버 배포 체크리스트
```
배포 전:
□ 모든 단위 테스트 통과
□ 커버리지 80% 이상
□ SonarQube 품질 게이트 통과
□ .env 운영 설정 확인 (SECRET_KEY, ENCRYPTION_KEY)
□ DB 백업 완료
배포 중:
□ 서비스 점검 공지
□ 현재 버전 백업
□ 새 버전 배포
□ DB 마이그레이션 (필요 시)
□ 서비스 재시작
배포 후:
□ /docs 페이지 접근 확인
□ 로그인 테스트
□ 주요 API 스모크 테스트
□ 스케줄러 동작 확인
```
---
## 11. 자주 하는 실수 및 주의사항
### 11.1 비동기 함수 실수
```python
# ❌ 실수: async 함수에서 동기 DB 호출
async def bad_function(db: AsyncSession):
result = db.execute(select(Item)) # ❌ await 누락
# ✅ 올바른 방법
async def good_function(db: AsyncSession):
result = await db.execute(select(Item)) # ✅
```
### 11.2 DB 세션 누수
```python
# ❌ 실수: 백그라운드 태스크에서 request 세션 사용
@router.post("/")
async def create(background_tasks: BackgroundTasks, db: AsyncSession = Depends(get_db)):
background_tasks.add_task(some_task, db) # ❌ 세션이 request 후 닫힘
# ✅ 올바른 방법: 백그라운드에서 새 세션 생성
async def some_task():
from database import SessionLocal
async with SessionLocal() as db: # ✅ 새 세션
pass
```
### 11.3 N+1 쿼리 문제
```python
# ❌ N+1 쿼리 발생
srs = (await db.execute(select(SRRequest))).scalars().all()
for sr in srs:
print(sr.work_logs) # 각 SR마다 추가 쿼리 발생!
# ✅ selectinload 사용
srs = (await db.execute(
select(SRRequest).options(selectinload(SRRequest.work_logs))
)).scalars().all()
```
### 11.4 환경 변수 검증
```python
# ✅ 앱 시작 시 필수 환경 변수 검증
class Settings(BaseSettings):
SECRET_KEY: str
ENCRYPTION_KEY: str
@validator("ENCRYPTION_KEY")
def encryption_key_must_be_64_hex(cls, v):
if len(v) != 64:
raise ValueError("ENCRYPTION_KEY는 64자리 hex 문자열이어야 합니다.")
return v
```
### 11.5 외부 API 완전 금지
```python
# ❌ 절대 금지: 외부 LLM/AI API 호출
import openai
client = openai.OpenAI(api_key="...") # ❌ 외부 API 사용 금지
# ✅ 내부 sLLM만 허용
async with httpx.AsyncClient() as client:
r = await client.post(f"{settings.SLLM_BASE_URL}/v1/chat/completions", ...)
```
---
*본 지침서를 준수하지 않은 코드는 코드 리뷰에서 반려될 수 있습니다.*
*보안 관련 규칙(6장)은 특히 엄격히 적용됩니다.*