### RFS Framework Development Setup Source: https://github.com/interactord/rfs-framework/blob/main/README.md Instructions for setting up the development environment for RFS Framework, including creating a virtual environment, installing development dependencies, and configuring pre-commit hooks. ```bash # 가상환경 생성 python -m venv venv source venv/bin/activate # Linux/Mac # 또는 venv\Scripts\activate # Windows # 개발용 의존성 설치 pip install -e ".[dev,test,docs]" # 프리커밋 훅 설정 pre-commit install ``` -------------------------------- ### RFS Framework E-commerce API Example Source: https://github.com/interactord/rfs-framework/blob/main/README.md A basic example of initializing an RFSApp for an e-commerce API, demonstrating the integration of core RFS components. ```Python from rfs_v4 import RFSApp from rfs_v4.core import Result from rfs_v4.state_machine import StateMachine from rfs_v4.reactive import Flux app = RFSApp() ``` -------------------------------- ### Install RFS Framework Dependencies Source: https://github.com/interactord/rfs-framework/blob/main/CLAUDE.md Installs the RFS framework with all features or specific feature sets using pip. This is the first step for setting up the development environment. ```Bash pip install -e ".[all]" pip install -e ".[web,database,test]" ``` -------------------------------- ### Install RFS Framework Source: https://github.com/interactord/rfs-framework/blob/main/README.md Installs the RFS Framework version 4 using pip. ```bash pip install rfs-v4 ``` -------------------------------- ### Setup Optimization Suite Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Configures an OptimizationSuite for performance analysis, specifying optimization types, target directories, and profiling duration. ```python from rfs_v4.optimization import ( PerformanceOptimizer, OptimizationSuite, OptimizationType ) optimizer = PerformanceOptimizer() suite = OptimizationSuite( types=[ OptimizationType.MEMORY, OptimizationType.CPU, OptimizationType.IO, OptimizationType.CLOUD_RUN ], target_directories=["./src"], profiling_duration=60 # 60초간 프로파일링 ) ``` -------------------------------- ### RFS CLI Project Management Source: https://github.com/interactord/rfs-framework/blob/main/README.md Provides examples of using the RFS CLI for project management tasks like creating projects, checking info, and managing dependencies. ```bash # 새 프로젝트 생성 rfs-cli create-project my-awesome-app --template fastapi # 프로젝트 정보 확인 rfs-cli project info # 의존성 관리 rfs-cli project deps --install ``` -------------------------------- ### Setup Validation Suite Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Defines a validation suite with specified categories, target directories, and configuration files for system validation. ```python from rfs_v4.validation import SystemValidator, ValidationSuite, ValidationCategory validator = SystemValidator() # 검증 스위트 정의 suite = ValidationSuite( categories=[ ValidationCategory.FUNCTIONAL, ValidationCategory.INTEGRATION, ValidationCategory.PERFORMANCE, ValidationCategory.SECURITY, ValidationCategory.COMPATIBILITY ], target_directories=["./src", "./tests"], config_files=["config.toml", "requirements.txt"] ) ``` -------------------------------- ### Docker Deployment for RFS Framework Source: https://github.com/interactord/rfs-framework/blob/main/README.md This Dockerfile sets up a Python 3.11 environment, copies project files, installs dependencies, exposes port 8080, and runs the RFS CLI server. ```dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . EXPOSE 8080 CMD ["rfs-cli", "serve", "--port", "8080"] ``` -------------------------------- ### Result Pattern Usage Examples Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Provides usage examples for the Result pattern's methods, illustrating how to create results, check their state, transform values using map and flat_map, and handle results with match. ```Python result = Result.success(42) print(result.value) # 42 result = Result.failure("Error message") print(result.error) # Error message result = Result.success(5).map(lambda x: x * 2) assert result.value == 10 def divide(x, y): if y == 0: return Result.failure("Division by zero") return Result.success(x / y) result = Result.success(10).flat_map(lambda x: divide(x, 2)) assert result.value == 5.0 result = Result.success("Hello") output = result.match( on_success=lambda x: f"Success: {x}", on_failure=lambda e: f"Error: {e}" ) assert output == "Success: Hello" ``` -------------------------------- ### Run RFS Development Server Source: https://github.com/interactord/rfs-framework/blob/main/CLAUDE.md Starts the RFS development server with automatic reloading on code changes and specifies the port. ```Bash rfs-cli dev --reload --port 8000 ``` -------------------------------- ### Performance Optimization Tips Source: https://github.com/interactord/rfs-framework/blob/main/README.md Python code examples demonstrating performance optimization techniques. This includes using streams for memory efficiency with batch processing, implementing caching for expensive operations, and utilizing asynchronous parallel processing. ```python # 1. 메모리 최적화를 위한 스트림 사용 async def process_large_dataset(): return await ( Flux.from_database("large_table") .buffer(100) # 배치 처리 .map(process_batch) .flat_map(lambda batch: Flux.from_iterable(batch)) .collect_list() .to_result() ) # 2. 캐싱으로 성능 향상 @app.cache(ttl=300) # 5분 캐시 async def expensive_operation() -> Result[str, str]: # 비용이 큰 연산 pass # 3. 비동기 병렬 처리 async def parallel_processing(): tasks = [ process_user(user_id) for user_id in user_ids ] results = await Flux.merge(*tasks).collect_list().to_result() return results ``` -------------------------------- ### Flux Creation Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Shows how to create Flux instances, which represent a reactive stream of multiple items. Examples include creating from iterables and generating sequences. ```Python from rfs_v4.reactive import Flux # 이터러블로 생성 flux = Flux.from_iterable([1, 2, 3, 4, 5]) # 범위 생성 range_flux = Flux.range(1, 10) ``` -------------------------------- ### Define and Add Custom Readiness Check Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Illustrates how to create a custom readiness check by subclassing `ReadinessCheck` and implementing the `execute` method. This example defines a `LoadTestCheck` to validate system performance under load, checking response time, error rate, and throughput. It also shows how to add this custom check to the checker. ```Python from rfs_v4.production import ReadinessCheck class LoadTestCheck(ReadinessCheck): async def execute(self) -> bool: # 부하 테스트 실행 result = await run_load_test( target_url="https://api.example.com", duration=60, concurrent_users=100 ) return ( result.avg_response_time < 200 and result.error_rate < 0.1 and result.throughput > 100 ) @property def name(self) -> str: return "Load Test Validation" @property def description(self) -> str: return "Validates system performance under load" @property def category(self) -> str: return "performance" # 커스텀 체크 추가 checker.add_check(LoadTestCheck()) ``` -------------------------------- ### Configuration Loading and Profiles Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Demonstrates loading application configurations from files (e.g., TOML) and managing environment-specific settings using profiles. ```Python from rfs_v4.core import ConfigManager # 설정 로드 config = ConfigManager.load("config.toml", MyAppConfig) # 환경별 설정 dev_config = ConfigManager.load_profile("development", MyAppConfig) prod_config = ConfigManager.load_profile("production", MyAppConfig) ``` -------------------------------- ### Contributing to RFS Framework Source: https://github.com/interactord/rfs-framework/blob/main/README.md Steps to contribute to the RFS Framework project, including forking the repository, setting up the development environment, running tests, and creating a pull request. ```bash # 1. 저장소 포크 git clone https://github.com/interactord/rfs-framework.git # 2. 개발 환경 설정 cd rfs-framework pip install -e ".[dev]" # 3. 테스트 실행 rfs-cli test --all # 4. PR 생성 git checkout -b feature/awesome-feature git commit -m "feat: add awesome feature" git push origin feature/awesome-feature ``` -------------------------------- ### RFS Result Pattern: Chaining vs Nested Ifs Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Compares two approaches for handling operations that can fail using the `Result` pattern. The 'good' example demonstrates chaining `flat_map` and `map` operations for a fluent and readable error-handling flow. The 'bad' example shows the less maintainable nested `if` statements typically used without such patterns. ```Python # ✅ 좋은 예: 체이닝 사용 async def process_user(user_id: int) -> Result[dict, str]: return await ( get_user(user_id) .flat_map(lambda user: validate_user(user)) .flat_map(lambda user: enrich_user_data(user)) .map(lambda user: format_user_response(user)) ) # ❌ 나쁜 예: 중첩된 if문 async def process_user_bad(user_id: int) -> Result[dict, str]: user_result = await get_user(user_id) if user_result.is_failure(): return user_result validation_result = await validate_user(user_result.value) if validation_result.is_failure(): return validation_result # ... 중첩 계속 ``` -------------------------------- ### Generate Documentation and Build RFS Project Source: https://github.com/interactord/rfs-framework/blob/main/CLAUDE.md Generates project documentation using the RFS CLI and builds the project for production deployment, specifically targeting Cloud Run. ```Bash # Generate documentation rfs-cli docs --all # Build for production rfs-cli build --platform cloud-run ``` -------------------------------- ### Custom Validation Rule Example Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Implements a custom validation rule 'DatabaseConnectionRule' to check database connectivity and adds it to the validator. ```python from rfs_v4.validation import ValidationRule class DatabaseConnectionRule(ValidationRule): async def check(self, context: dict) -> bool: try: db = context.get("database") await db.ping() return True except Exception: return False @property def name(self) -> str: return "Database Connection Check" @property def category(self) -> ValidationCategory: return ValidationCategory.FUNCTIONAL # 커스텀 룰 등록 validator.add_rule(DatabaseConnectionRule()) ``` -------------------------------- ### RFS Framework Configuration Management Source: https://github.com/interactord/rfs-framework/blob/main/README.md Shows how to load and use configuration settings with the RFS Framework, including environment-specific configurations. ```Python from rfs import RFSConfig, get_config # 설정 파일 로드 (config.toml) config = get_config() print(f"애플리케이션 환경: {config.environment}") print(f"데이터베이스 URL: {config.database.url}") # 환경별 설정 사용 if config.environment == "production": print("프로덕션 환경에서 실행 중") else: print("개발 환경에서 실행 중") ``` -------------------------------- ### RFS CLI Deployment Commands Source: https://github.com/interactord/rfs-framework/blob/main/README.md Shows RFS CLI commands for deploying applications to Cloud Run, checking deployment status, and viewing logs. ```bash # Cloud Run 배포 rfs-cli deploy cloud-run --region asia-northeast3 # 배포 상태 확인 rfs-cli deploy status # 로그 확인 rfs-cli deploy logs --follow ``` -------------------------------- ### RFS CLI Development Commands Source: https://github.com/interactord/rfs-framework/blob/main/README.md Demonstrates RFS CLI commands for development, including running a dev server and performing code quality checks. ```bash # 개발 서버 실행 rfs-cli dev --reload --port 8000 # 코드 품질 검사 rfs-cli dev lint rfs-cli dev test rfs-cli dev security-scan ``` -------------------------------- ### Execute Deployment with Canary Strategy Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Shows how to use the `ProductionDeployer` to execute a deployment with a specified strategy, such as CANARY. This includes configuring canary parameters like percentage, validation duration, and rollback threshold. It also demonstrates how to monitor the deployment status after initiation. ```Python from rfs_v4.production import ProductionDeployer, DeploymentStrategy deployer = ProductionDeployer() # 카나리 배포 전략 strategy = DeploymentStrategy.CANARY( canary_percentage=10, validation_duration=300, # 5분 rollback_threshold=0.05 # 5% 에러율 ) # 배포 실행 deployment_result = await deployer.deploy( version="v1.2.0", strategy=strategy, health_check_url="/health", readiness_check_url="/ready" ) if deployment_result.is_success(): deployment_id = deployment_result.value["deployment_id"] # 배포 상태 모니터링 status = await deployer.get_deployment_status(deployment_id) print(f"Deployment Status: {status}") ``` -------------------------------- ### RFS Error Handling with Result Type Source: https://github.com/interactord/rfs-framework/blob/main/CLAUDE.md Demonstrates the RFS framework's philosophy of never throwing exceptions in business logic. Instead, it uses the Result type for explicit success or failure handling, illustrated with a division example. ```Python from rfs.core import Result, Success, Failure def divide(a: int, b: int) -> Result[float, str]: if b == 0: return Failure("Division by zero") return Success(a / b) ``` -------------------------------- ### Cloud Task Queue Definition Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Provides an example of defining a task for asynchronous processing using Cloud Task Queue. This involves creating a CloudTaskQueue instance and defining a task handler function with a specific task type. ```Python from rfs_v4.cloud_run import CloudTaskQueue, TaskDefinition, task_handler queue = CloudTaskQueue( project_id="my-project", location="us-central1", queue_name="default-queue" ) # 태스크 정의 @task_handler("send-email") async def send_email_task(payload: dict): email = payload["email"] subject = payload["subject"] body = payload["body"] await email_service.send(email, subject, body) return {"status": "sent", "email": email} ``` -------------------------------- ### Deploy RFS to Cloud Run Source: https://github.com/interactord/rfs-framework/blob/main/CLAUDE.md Deploys the RFS project to Google Cloud Run, including commands to check deployment status and view logs. ```Bash # Deploy to Cloud Run rfs-cli deploy cloud-run --region asia-northeast3 # Check deployment status rfs-cli deploy status # View logs rfs-cli deploy logs --follow ``` -------------------------------- ### Get Order Items API Endpoint Source: https://github.com/interactord/rfs-framework/blob/main/README.md An asynchronous API endpoint to retrieve order items. It fetches items from a database using a reactive stream, maps item data, filters items with quantity greater than 0, and collects them into a list. ```python @app.route("/orders/{order_id}/items") async def get_order_items(order_id: str) -> Result[list, str]: # 반응형 스트림으로 주문 아이템 처리 items = await ( Flux.from_database(f"orders/{order_id}/items") .map(lambda item: { "id": item.id, "name": item.name, "price": item.price, "quantity": item.quantity }) .filter(lambda item: item["quantity"] > 0) .collect_list() .to_result() ) return items ``` -------------------------------- ### Create RFS Distribution Packages Source: https://github.com/interactord/rfs-framework/blob/main/CLAUDE.md Creates distribution packages for the RFS project using the standard Python build process. ```Bash python -m build ``` -------------------------------- ### RFS Framework v3 코드 분석 Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md 이전 버전(v3)의 RFS Framework를 사용하는 코드를 분석하는 데 사용되는 Bash 스크립트입니다. RFS 관련 import 문, 동기 함수, 예외 처리 패턴을 찾는 데 도움이 됩니다. ```bash # v3 사용 패턴 분석 grep -r "from rfs" . --include="*.py" grep -r "import rfs" . --include="*.py" # 동기 함수 식별 grep -r "def " . --include="*.py" | grep -v "async def" # 예외 처리 패턴 식별 grep -r "try:" . --include="*.py" grep -r "except" . --include="*.py" ``` -------------------------------- ### RFS Framework Protocol Support for Serialization Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Illustrates how to use Python's `Protocol` for defining interfaces, specifically for serialization. This example defines a `Serializable` protocol with a `to_dict` method and shows a function `serialize_items` that accepts a list of `Serializable` objects and returns them as a Mono of dictionaries. ```Python from typing import Protocol from rfs_v4.reactive import Mono class Serializable(Protocol): def to_dict(self) -> dict: ... def serialize_items(items: list[Serializable]) -> Mono[list[dict]]: return Mono.just([item.to_dict() for item in items]) ``` -------------------------------- ### RFS Framework v4 마이그레이션 환경 준비 Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md RFS Framework 4.0으로 마이그레이션하기 위한 환경 준비 단계입니다. Python 버전 확인, 현재 의존성 백업, 새 가상환경 생성 및 RFS Framework 설치를 포함합니다. ```bash # Python 버전 확인 (3.10+ 필수) python --version # 현재 의존성 백업 pip freeze > requirements_v3_backup.txt # 새 가상환경 생성 python -m venv venv_v4 source venv_v4/bin/activate # RFS Framework 설치 pip install rfs-framework ``` -------------------------------- ### RFS Framework Basic Usage Source: https://github.com/interactord/rfs-framework/blob/main/README.md Demonstrates basic usage of the RFS Framework, including the Result pattern for error handling and system validation. ```Python from rfs import Result, Success, Failure from rfs import SystemValidator, PerformanceOptimizer # Result 패턴으로 안전한 에러 핸들링 def divide(a: int, b: int) -> Result[float, str]: if b == 0: return Failure("0으로 나눌 수 없습니다") return Success(a / b) # 결과 처리 result = divide(10, 2) if result.is_success: print(f"결과: {result.unwrap()}") # 결과: 5.0 else: print(f"오류: {result.unwrap_err()}") # 시스템 검증 사용 validator = SystemValidator() validation_result = validator.validate_system() print(f"시스템 상태: {'정상' if validation_result.is_valid else '문제 발견'}") ``` -------------------------------- ### RFS Framework v3에서 v4로 의존성 업데이트 Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md requirements.txt 파일에서 RFS Framework 및 관련 라이브러리의 의존성을 이전 버전(v3.x)에서 새 버전(v4.0)으로 업데이트하는 방법을 보여주는 diff 형식의 코드입니다. ```diff # Before (v3.x) - rfs-framework==3.5.2 - python-dotenv==0.19.0 - pydantic==1.10.2 # After (v4.0) + rfs-framework-v4==4.0.0 + python-dotenv>=1.0.0 + pydantic>=2.5.0 ``` -------------------------------- ### Configuration Management with RFSConfig Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Shows how to define application configuration using RFSConfig, including specifying typed attributes, default values, and environment-specific prefixes. ```Python from rfs_v4.core import RFSConfig, Environment class MyAppConfig(RFSConfig): database_url: str redis_url: str debug: bool = False class Config: env_prefix = "MYAPP_" ``` -------------------------------- ### Configuration File Format Migration (YAML to TOML) Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md Demonstrates the conversion of configuration settings from YAML format to TOML format, including the structure for database, logging, and cache settings, as well as environment-specific configurations. ```YAML database: host: localhost port: 5432 name: myapp logging: level: INFO format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s" cache: redis_url: redis://localhost:6379 ttl: 3600 ``` -------------------------------- ### Cloud Run Deployment Commands Source: https://github.com/interactord/rfs-framework/blob/main/README.md Bash commands for deploying an application to Google Cloud Run using the RFS CLI. This includes building the project, deploying to a specific region with resource configurations, and mapping a custom domain. ```bash # 1. 프로젝트 빌드 rfs-cli build --platform cloud-run # 2. 배포 rfs-cli deploy cloud-run \ --region asia-northeast3 \ --memory 1Gi \ --cpu 2 \ --max-instances 100 # 3. 도메인 매핑 rfs-cli deploy domain --name api.example.com ``` -------------------------------- ### Configuration File Format Migration (YAML to TOML) Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md Demonstrates the conversion of configuration settings from YAML format to TOML format, including the structure for database, logging, and cache settings, as well as environment-specific configurations. ```TOML [database] host = "localhost" port = 5432 name = "myapp" [logging] level = "INFO" format = "% (asctime)s - % (name)s - % (levelname)s - % (message)s" [cache] redis_url = "redis://localhost:6379" ttl = 3600 # 환경별 설정 [development] extends = "base" log_level = "DEBUG" [production] extends = "base" log_level = "INFO" ``` -------------------------------- ### Environment Configuration (config.toml) Source: https://github.com/interactord/rfs-framework/blob/main/README.md TOML configuration file defining different environment settings for development, production, and cloud run deployments. It includes database URLs, Redis URLs, log levels, and allows extending configurations. ```toml # config.toml [development] database_url = "sqlite:///dev.db" redis_url = "redis://localhost:6379" log_level = "DEBUG" [production] database_url = "${DATABASE_URL}" redis_url = "${REDIS_URL}" log_level = "INFO" [cloud_run] extends = "production" port = 8080 workers = 4 ``` -------------------------------- ### Application Configuration Loading Source: https://github.com/interactord/rfs-framework/blob/main/README.md Python code for loading application configuration from a TOML file using the RFS Config class. It demonstrates how to access configuration profiles like production and development. ```python from rfs_v4.core import Config, ConfigProfile config = Config.load("config.toml") # 환경별 설정 로드 if config.profile == ConfigProfile.PRODUCTION: # 프로덕션 설정 pass elif config.profile == ConfigProfile.DEVELOPMENT: # 개발 설정 pass ``` -------------------------------- ### RFS Framework v4 프로젝트 설정 (pyproject.toml) Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md RFS Framework 4.0 환경에서 프로젝트 설정을 위해 pyproject.toml 파일을 사용하는 예시입니다. 프로젝트 이름, 버전, 그리고 RFS Framework v4 및 Pydantic의 의존성을 명시합니다. ```toml [project] name = "my-project" version = "1.0.0" dependencies = [ "rfs-framework-v4==4.0.0", "pydantic>=2.5.0", ] ``` -------------------------------- ### Environment Detection Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Illustrates how to detect the current runtime environment (e.g., development, production) to apply appropriate configurations. ```Python from rfs_v4.core import detect_current_environment, Environment env = detect_current_environment() if env == Environment.PRODUCTION: # 프로덕션 설정 pass elif env == Environment.DEVELOPMENT: # 개발 설정 pass ``` -------------------------------- ### Integration Testing Commands Source: https://github.com/interactord/rfs-framework/blob/main/README.md Bash commands for running integration tests using the RFS CLI. It shows how to run all integration tests or tests for specific modules. ```bash # CLI를 통한 통합 테스트 rfs-cli test --integration # 특정 모듈 테스트 rfs-cli test --module core rfs-cli test --module reactive ``` -------------------------------- ### Run Production Readiness Check Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Demonstrates how to use the ProductionReadinessChecker to perform a readiness check against a target level, such as PRODUCTION_READY. It shows how to interpret the results, including overall score, readiness status, and category-specific checks, highlighting high-severity failures. ```Python from rfs_v4.production import ProductionReadinessChecker, ReadinessLevel checker = ProductionReadinessChecker() # 프로덕션 준비성 검사 readiness_result = await checker.run_readiness_check( target_level=ReadinessLevel.PRODUCTION_READY ) if readiness_result.is_success(): report = readiness_result.value print(f"Overall Readiness: {report.overall_score}/100") print(f"Ready for Production: {'Yes' if report.is_ready else 'No'}") # 카테고리별 점검 결과 for category, checks in report.category_results.items(): passed = sum(1 for c in checks if c.passed) total = len(checks) print(f"{category}: {passed}/{total} passed") # 실패한 검사 표시 failed = [c for c in checks if not c.passed and c.severity == "high"] for check in failed: print(f" ❌ {check.check_name}: {check.message}") ``` -------------------------------- ### Code Quality and Linting with RFS CLI Source: https://github.com/interactord/rfs-framework/blob/main/CLAUDE.md Applies code formatting with black, sorts imports with isort, performs type checking with mypy, and runs security scanning with bandit. The RFS CLI also provides integrated commands for linting and security scanning. ```Bash # Format code with black black src/rfs # Sort imports isort src/rfs # Type checking mypy src/rfs # Security scanning bandit -r src/rfs # All linting/validation via CLI rfs-cli dev lint rfs-cli dev security-scan ``` -------------------------------- ### RFS Framework v3에서 v4로 Import 경로 변경 Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md RFS Framework 마이그레이션 시 import 경로 변경 예시입니다. 이전 버전(v3.x)의 import 문을 새 버전(v4.0)의 import 문으로 어떻게 변경해야 하는지 보여줍니다. ```python # Before (v3.x) - 예외 기반 def get_user(user_id: int) -> dict: if user_id < 1: raise ValueError("Invalid user ID") user = database.get_user(user_id) if not user: raise NotFoundError("User not found") return user # After (v4.0) - Result 패턴 async def get_user(user_id: int) -> Result[dict, str]: if user_id < 1: return Result.failure("Invalid user ID") user_result = await database.get_user(user_id) if user_result.is_failure(): return Result.failure("User not found") return Result.success(user_result.value) ``` ```python # v3.x → v4.0 Import 매핑표 v3_imports = { "rfs.core.BaseService": "rfs_v4.core.Result", "rfs.config.Config": "rfs_v4.core.Config", "rfs.async_utils": "rfs_v4.reactive", "rfs.state": "rfs_v4.state_machine", "rfs.events": "rfs_v4.events", } ``` -------------------------------- ### RFS Framework State Machine Source: https://github.com/interactord/rfs-framework/blob/main/README.md Illustrates the use of the StateMachine component in the RFS Framework for managing application states. ```Python from rfs import StateMachine, State, Transition from rfs import Result # 간단한 주문 상태 머신 order_machine = StateMachine( initial_state="pending", states=["pending", "processing", "completed", "cancelled"] ) # 상태 전환 print(f"현재 상태: {order_machine.current_state}") # pending order_machine.transition_to("processing") print(f"변경된 상태: {order_machine.current_state}") # processing ``` -------------------------------- ### RFS Framework v4 Result 패턴 적용 예시 Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md RFS Framework 4.0에서 Result 패턴을 사용하여 에러를 처리하는 방법을 보여주는 Python 코드입니다. 이전 버전의 예외 기반 처리 방식과 비교하여 Result.success 및 Result.failure 사용법을 설명합니다. ```python # Before (v3.x) def process_order(order_data: dict) -> dict: try: # 주문 검증 if not order_data.get('items'): raise ValueError("No items in order") # 재고 확인 for item in order_data['items']: stock = inventory.check_stock(item['id']) if stock < item['quantity']: raise InsufficientStockError(f"Not enough stock for {item['name']}") # 주문 생성 order = Order.create(order_data) return order.to_dict() except ValueError as e: raise BadRequestError(str(e)) except InsufficientStockError as e: raise ConflictError(str(e)) # After (v4.0) async def process_order(order_data: dict) -> Result[dict, str]: # 주문 검증 if not order_data.get('items'): return Result.failure("No items in order") # 재고 확인을 반응형 스트림으로 처리 stock_check = await ( Flux.from_iterable(order_data['items']) .flat_map(lambda item: check_item_stock(item)) .collect_list() .to_result() ) if stock_check.is_failure(): return stock_check # 주문 생성 order_result = await Order.create_async(order_data) return order_result.map(lambda order: order.to_dict()) async def check_item_stock(item: dict) -> Result[dict, str]: stock_result = await inventory.check_stock_async(item['id']) if stock_result.is_failure(): return stock_result if stock_result.value < item['quantity']: return Result.failure(f"Not enough stock for {item['name']}") return Result.success(item) ``` -------------------------------- ### Test Execution and Coverage Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md Commands to run all tests, measure code coverage, and generate a coverage report. ```Bash # 모든 테스트 실행 pytest --cov=src tests/ # 커버리지 확인 (90% 이상 권장) pytest --cov=src --cov-report=html tests/ ``` -------------------------------- ### Performance Benchmarking Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md Compares the performance of old and new implementations by measuring execution time and calculating the percentage improvement. ```Python import time import asyncio async def benchmark_old_vs_new(): # 기존 방식 start = time.time() old_result = process_users_sync(user_ids) old_time = time.time() - start # 새 방식 start = time.time() new_result = await process_users_async(user_ids) new_time = time.time() - start print(f"Old: {old_time:.3f}s, New: {new_time:.3f}s") print(f"Improvement: {((old_time - new_time) / old_time * 100):.1f}%") ``` -------------------------------- ### Cloud Run Service Discovery with Load Balancing Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Explains how to perform service calls with load balancing enabled using Cloud Run's service discovery. It specifically shows how to configure round-robin load balancing for requests. ```Python # 라운드 로빈 로드 밸런싱 response = await discovery.call_service( service_name="user-service", method="POST", path="/users", json={"name": "John", "email": "john@example.com"}, load_balancing="round_robin" ) ``` -------------------------------- ### Cloud Run Service Discovery and Call Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Demonstrates how to discover registered services and make calls to them using Cloud Run's service discovery. It includes finding services by name and invoking them with specific HTTP methods, paths, and request bodies. ```Python # 서비스 조회 services = await discovery.discover_services("user-service") user_service = services[0] # 서비스 호출 response = await discovery.call_service( service_name="user-service", method="GET", path="/users/123" ) ``` -------------------------------- ### Service Registry Usage Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Explains how to use the ServiceRegistry for dependency injection, including registering stateless services and retrieving them. ```Python from rfs_v4.core import ServiceRegistry, stateless registry = ServiceRegistry() # 서비스 등록 @stateless class DatabaseService: def get_connection(self): return "database connection" registry.register(DatabaseService) # 서비스 조회 db_service = registry.get(DatabaseService) ``` -------------------------------- ### Security Features and Practices Source: https://github.com/interactord/rfs-framework/blob/main/README.md Python code demonstrating security practices, including vulnerability scanning using SecurityScanner, data encryption and decryption with AES-256, and input validation using Pydantic schemas with the @validate_input decorator. ```python from rfs_v4.security import SecurityScanner, encrypt, decrypt # 보안 스캔 scanner = SecurityScanner() vulnerabilities = await scanner.scan_directory("./src") # 데이터 암호화 encrypted_data = encrypt("sensitive information", key) decrypted_data = decrypt(encrypted_data, key) # 입력 검증 @app.route("/api/users") @validate_input(UserCreateSchema) async def create_user(data: dict) -> Result[dict, str]: # 자동으로 검증된 데이터 pass ``` -------------------------------- ### Monitoring Performance and Errors Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md Demonstrates how to use a MetricsCollector to track performance and errors for critical functions. ```Python from rfs_v4.monitoring import MetricsCollector metrics = MetricsCollector() # 비즈니스 메트릭 추가 @metrics.track_performance async def critical_business_function(): pass # 에러율 모니터링 @metrics.track_errors async def error_prone_function(): pass ``` -------------------------------- ### Python Testing Migration: unittest to pytest Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md Illustrates the migration of Python unit tests from the standard 'unittest' framework to 'pytest'. This includes changes in test structure, fixture usage, and asynchronous test handling with 'asyncio' and 'AsyncMock'. ```Python # Before (v3.x) import unittest from unittest.mock import Mock, patch class TestUserService(unittest.TestCase): def setUp(self): self.service = UserService() def test_get_user_success(self): user = self.service.get_user(1) self.assertIsInstance(user, dict) self.assertEqual(user['id'], 1) def test_get_user_not_found(self): with self.assertRaises(NotFoundError): self.service.get_user(999) # After (v4.0) import pytest import asyncio from unittest.mock import AsyncMock class TestUserService: @pytest.fixture def service(self): mock_db = AsyncMock() return UserService(mock_db) async def test_get_user_success(self, service): service.db.query_one.return_value = {"id": 1, "name": "John"} result = await service.get_user(1) assert result.is_success() assert result.value['id'] == 1 async def test_get_user_not_found(self, service): service.db.query_one.return_value = None result = await service.get_user(999) assert result.is_failure() assert "not found" in result.error ``` -------------------------------- ### Configuration Loading Mechanism Change Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md Illustrates the change in how configuration files are loaded and accessed in RFS Framework, moving from v3.x's direct attribute access to v4.0's section-based dictionary access. ```Python # Before (v3.x) from rfs.config import Config config = Config.load("config.yaml") db_host = config.database.host ``` ```Python # After (v4.0) from rfs_v4.core import Config, ConfigProfile config = Config.load("config.toml") db_config = config.get_section("database") db_host = db_config["host"] ``` -------------------------------- ### Type Annotation Compatibility Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md Shows how to correctly use type hints for compatibility across different Python versions, especially for union types. ```Python # Python 3.10+ 문법 사용 from typing import Union # 잘못된 사용 (Python 3.8) def process(data: dict | None) -> str | None: pass # 올바른 사용 (Python 3.10+) def process(data: dict | None) -> str | None: pass # 또는 하위 호환성을 위해 def process(data: Union[dict, None]) -> Union[str, None]: pass ``` -------------------------------- ### Mono Creation Source: https://github.com/interactord/rfs-framework/blob/main/API_REFERENCE.md Shows various ways to create Mono instances, which represent a reactive stream of 0 or 1 item. This includes creating from a value, an empty stream, an error, or lazily. ```Python from rfs_v4.reactive import Mono # 값으로 생성 mono = Mono.just("Hello") # 빈 스트림 empty = Mono.empty() # 에러 스트림 error = Mono.error(Exception("Something went wrong")) # 지연 생성 lazy = Mono.from_callable(lambda: expensive_operation()) ``` -------------------------------- ### Python Order State Management: If/Else vs. State Machine Source: https://github.com/interactord/rfs-framework/blob/main/MIGRATION_GUIDE.md Demonstrates the evolution of order status management from a simple if/else conditional structure in v3.x to a more robust state machine implementation in v4.0 using the 'rfs_v4.state_machine' library. The state machine approach enhances maintainability and prevents invalid state transitions. ```Python # Before (v3.x) - if/else 기반 class Order: def __init__(self, data): self.status = "pending" self.data = data def process_payment(self): if self.status != "pending": raise InvalidStateError("Cannot process payment") # 결제 처리 로직 self.status = "paid" def ship_order(self): if self.status != "paid": raise InvalidStateError("Cannot ship unpaid order") # 배송 처리 로직 self.status = "shipped" # After (v4.0) - 상태 머신 from rfs_v4.state_machine import StateMachine, State, Transition class OrderStateMachine: def __init__(self): self.machine = StateMachine( initial_state=State("pending"), states=[ State("pending"), State("paid"), State("shipped"), State("delivered"), State("cancelled") ], transitions=[ Transition("pending", "pay", "paid"), Transition("paid", "ship", "shipped"), Transition("shipped", "deliver", "delivered"), Transition("pending", "cancel", "cancelled"), Transition("paid", "cancel", "cancelled") ] ) async def process_payment(self, order_id: str) -> Result[str, str]: # 결제 로직 payment_result = await self._process_payment_logic(order_id) if payment_result.is_failure(): return payment_result # 상태 전환 transition_result = await self.machine.dispatch("pay") if transition_result.is_failure(): return Result.failure("Invalid state for payment") return Result.success("Payment processed successfully") ```