Где хранил бизнес-логику в своих проектах?

Организация бизнес-логики в проектах

За 10+ лет разработки я применял различные подходы к организации бизнес-логики, эволюционируя от монолитных структур к чистой архитектуре. Расскажу о моих принципах и практиках.

Слоистая архитектура (Layered Architecture)

В большинстве production проектов я использую классическую трёхслойную архитектуру:

presentation/ → application/ → domain/

Domain слой — это сердце: бизнес-сущности, value objects и интерфейсы репозиториев. Здесь НЕ должно быть зависимостей от фреймворков.

# domain/models.py
class User:
    def __init__(self, id: UUID, email: str, password_hash: str):
        self.id = id
        self.email = email
        self.password_hash = password_hash
    
    def verify_password(self, password: str) -> bool:
        return bcrypt.verify(password, self.password_hash)

class UserRepository(ABC):
    @abstractmethod
    async def find_by_id(self, id: UUID) -> User | None:
        pass

Application слой — use cases и сервисы. Здесь располагается орхестрация бизнес-процессов. Зависит только от domain.

# application/services.py
class RegisterUserService:
    def __init__(self, user_repo: UserRepository, email_service: EmailService):
        self.user_repo = user_repo
        self.email_service = email_service
    
    async def execute(self, email: str, password: str) -> User:
        if await self.user_repo.find_by_email(email):
            raise UserAlreadyExists()
        
        user = User(id=uuid4(), email=email, password_hash=hash_password(password))
        await self.user_repo.save(user)
        await self.email_service.send_confirmation(user.email)
        return user

Infrastructure слой — реализация репозиториев, работа с БД, внешними API. Зависит от application и domain.

# infrastructure/repositories.py
class PostgresUserRepository(UserRepository):
    def __init__(self, session: AsyncSession):
        self.session = session
    
    async def find_by_id(self, id: UUID) -> User | None:
        result = await self.session.execute(
            select(UserModel).where(UserModel.id == id)
        )
        return result.scalar_one_or_none()

Presentation слой (FastAPI, Flask, CLI) — HTTP endpoints, CLI команды. Вызывает use cases из application.

Критические принципы

Зависимости текут внутрь: presentation → application → domain. Никогда наоборот.

Бизнес-логика не в endpoints: Controllers должны быть тонкие — парсят request, вызывают service, возвращают response.

# ❌ НЕПРАВИЛЬНО
@app.post("/users")
async def register(email: str, password: str, db: Session):
    if await db.query(User).filter_by(email=email).first():
        raise HTTPException(400)
    user = User(email=email, password=hash_password(password))
    db.add(user)
    await db.commit()
    return {"id": user.id}

# ✅ ПРАВИЛЬНО
@app.post("/users")
async def register(dto: RegisterRequest, service: RegisterUserService):
    user = await service.execute(dto.email, dto.password)
    return UserResponse.from_domain(user)

DDD и Bounded Contexts

В сложных системах использую Domain-Driven Design: разбиваю систему на bounded contexts.

# contexts/orders/domain/models.py
class Order(AggregateRoot):
    def __init__(self, id: UUID, user_id: UUID):
        self.id = id
        self.user_id = user_id
        self.items: List[OrderItem] = []
        self.status = OrderStatus.PENDING
        self.events: List[DomainEvent] = []
    
    def add_item(self, product_id: UUID, quantity: int):
        if self.status != OrderStatus.PENDING:
            raise OrderAlreadyProcessed()
        item = OrderItem(product_id=product_id, quantity=quantity)
        self.items.append(item)
        self.events.append(ItemAddedEvent(order_id=self.id, item=item))

Ключевые подходы

• Изоляция от инфраструктуры: бизнес-логика в domain и application
• Тестируемость: сервисы не зависят от БД
• Масштабируемость: слои разделены по ответственности
• Документирование: архитектурные решения в docs/