Как спроектируешь ООП для файлообменника?

Проектирование ООП архитектуры файлообменника

Требования к системе

Файлообменник должен поддерживать:

• загрузку и скачивание файлов
• управление пользователями и правами доступа
• версионирование и истории файлов
• шифрование и безопасность
• масштабируемость и производительность

Архитектурные слои

Использую onion architecture (чистая архитектура):

presentation/ → application/ → domain/ → infrastructure/
  (API)        (use cases)   (entities) (DB, storage)

Domain Layer — основные сущности

from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from uuid import UUID

class AccessLevel(Enum):
    VIEW = 'view'
    EDIT = 'edit'
    ADMIN = 'admin'

@dataclass
class User:
    id: UUID
    username: str
    email: str
    created_at: datetime
    
    def can_access(self, resource_id: UUID, level: AccessLevel) -> bool:
        # Логика проверки доступа
        pass

@dataclass
class FileMetadata:
    id: UUID
    owner_id: UUID
    name: str
    size: int
    mime_type: str
    created_at: datetime
    updated_at: datetime
    is_deleted: bool = False
    
    def is_valid_name(self) -> bool:
        return len(self.name) > 0 and len(self.name) <= 255

@dataclass
class FileShare:
    id: UUID
    file_id: UUID
    user_id: UUID
    access_level: AccessLevel
    created_at: datetime
    expires_at: datetime | None
    
    def is_expired(self) -> bool:
        if not self.expires_at:
            return False
        return datetime.now() > self.expires_at

@dataclass
class FileVersion:
    id: UUID
    file_id: UUID
    version_number: int
    storage_path: str
    created_at: datetime
    created_by: UUID
    checksum: str  # SHA256

Domain Services

class FileAccessService:
    """Бизнес-логика проверки доступа"""
    
    def __init__(self, share_repo):
        self.share_repo = share_repo
    
    def can_view_file(self, user_id: UUID, file_id: UUID) -> bool:
        share = self.share_repo.find_by_user_and_file(
            user_id, file_id
        )
        if not share:
            return False
        return not share.is_expired()
    
    def get_access_level(
        self, 
        user_id: UUID, 
        file_id: UUID
    ) -> AccessLevel | None:
        share = self.share_repo.find_by_user_and_file(
            user_id, file_id
        )
        return share.access_level if share else None

class FileStorageService:
    """Бизнес-логика хранения файлов"""
    
    def calculate_checksum(self, content: bytes) -> str:
        import hashlib
        return hashlib.sha256(content).hexdigest()
    
    def validate_file(
        self, 
        metadata: FileMetadata, 
        content: bytes
    ) -> bool:
        # Проверка размера, типа, вируса
        MAX_SIZE = 5 * 1024 * 1024 * 1024  # 5GB
        if metadata.size > MAX_SIZE:
            return False
        return metadata.is_valid_name()

Application Layer — Use Cases

class UploadFileUseCase:
    def __init__(
        self,
        file_repo,
        file_storage: FileStorageService,
        access_service: FileAccessService
    ):
        self.file_repo = file_repo
        self.file_storage = file_storage
        self.access_service = access_service
    
    def execute(
        self,
        user_id: UUID,
        filename: str,
        content: bytes,
        mime_type: str
    ) -> FileMetadata:
        # Валидация
        metadata = FileMetadata(
            id=UUID(),
            owner_id=user_id,
            name=filename,
            size=len(content),
            mime_type=mime_type,
            created_at=datetime.now(),
            updated_at=datetime.now()
        )
        
        if not self.file_storage.validate_file(metadata, content):
            raise ValueError('Invalid file')
        
        # Сохранение
        checksum = self.file_storage.calculate_checksum(content)
        storage_path = self.file_storage.save(
            metadata.id, content
        )
        
        # Создание версии
        version = FileVersion(
            id=UUID(),
            file_id=metadata.id,
            version_number=1,
            storage_path=storage_path,
            created_at=datetime.now(),
            created_by=user_id,
            checksum=checksum
        )
        
        # Сохранение в БД
        self.file_repo.save(metadata)
        self.file_repo.save_version(version)
        
        return metadata

class DownloadFileUseCase:
    def __init__(self, file_repo, access_service):
        self.file_repo = file_repo
        self.access_service = access_service
    
    def execute(
        self,
        user_id: UUID,
        file_id: UUID
    ) -> bytes:
        # Проверка доступа
        if not self.access_service.can_view_file(user_id, file_id):
            raise PermissionError('No access')
        
        # Получение файла
        metadata = self.file_repo.find_by_id(file_id)
        if not metadata or metadata.is_deleted:
            raise FileNotFoundError()
        
        # Загрузка из хранилища
        content = self.file_repo.load(metadata.id)
        return content

class ShareFileUseCase:
    def __init__(self, share_repo, file_repo, access_service):
        self.share_repo = share_repo
        self.file_repo = file_repo
        self.access_service = access_service
    
    def execute(
        self,
        owner_id: UUID,
        file_id: UUID,
        target_user_id: UUID,
        access_level: AccessLevel,
        expires_in_days: int | None = None
    ) -> FileShare:
        # Проверка, что владелец
        file_meta = self.file_repo.find_by_id(file_id)
        if file_meta.owner_id != owner_id:
            raise PermissionError('Not owner')
        
        # Создание share
        from datetime import timedelta
        expires_at = None
        if expires_in_days:
            expires_at = datetime.now() + timedelta(days=expires_in_days)
        
        share = FileShare(
            id=UUID(),
            file_id=file_id,
            user_id=target_user_id,
            access_level=access_level,
            created_at=datetime.now(),
            expires_at=expires_at
        )
        
        self.share_repo.save(share)
        return share

Infrastructure Layer — Repositories

from abc import ABC, abstractmethod

class FileRepository(ABC):
    @abstractmethod
    def save(self, metadata: FileMetadata) -> None:
        pass
    
    @abstractmethod
    def find_by_id(self, file_id: UUID) -> FileMetadata | None:
        pass
    
    @abstractmethod
    def save_version(self, version: FileVersion) -> None:
        pass

class SQLAlchemyFileRepository(FileRepository):
    def __init__(self, session):
        self.session = session
    
    def save(self, metadata: FileMetadata) -> None:
        file_model = FileModel(
            id=metadata.id,
            owner_id=metadata.owner_id,
            name=metadata.name,
            size=metadata.size,
            mime_type=metadata.mime_type,
            created_at=metadata.created_at,
            updated_at=metadata.updated_at,
            is_deleted=metadata.is_deleted
        )
        self.session.add(file_model)
        self.session.commit()

class S3FileStorage:
    """Хранение файлов в S3"""
    
    def __init__(self, bucket_name: str):
        import boto3
        self.s3 = boto3.client('s3')
        self.bucket = bucket_name
    
    def save(self, file_id: UUID, content: bytes) -> str:
        key = f'files/{file_id}'
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=content
        )
        return key
    
    def load(self, file_id: UUID) -> bytes:
        key = f'files/{file_id}'
        response = self.s3.get_object(Bucket=self.bucket, Key=key)
        return response['Body'].read()

Presentation Layer — API

from fastapi import APIRouter, HTTPException, Depends

router = APIRouter(prefix='/api/v1/files')

@router.post('/upload')
async def upload_file(
    filename: str,
    content: bytes,
    current_user: User = Depends(get_current_user),
    use_case: UploadFileUseCase = Depends()
):
    try:
        metadata = use_case.execute(
            current_user.id,
            filename,
            content,
            'application/octet-stream'
        )
        return {'file_id': metadata.id}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

@router.get('/{file_id}/download')
async def download_file(
    file_id: UUID,
    current_user: User = Depends(get_current_user),
    use_case: DownloadFileUseCase = Depends()
):
    try:
        content = use_case.execute(current_user.id, file_id)
        return {'content': content}
    except PermissionError:
        raise HTTPException(status_code=403)

@router.post('/{file_id}/share')
async def share_file(
    file_id: UUID,
    target_user_id: UUID,
    access_level: AccessLevel,
    current_user: User = Depends(get_current_user),
    use_case: ShareFileUseCase = Depends()
):
    share = use_case.execute(
        current_user.id,
        file_id,
        target_user_id,
        access_level
    )
    return {'share_id': share.id}

Ключевые принципы

• DDD (Domain-Driven Design): Бизнес-логика в domain layer
• SOLID: Каждый класс отвечает за одно
• Инверсия зависимостей: Repositories через DI
• Разделение ответственности: API не знает о БД
• Тестируемость: Mock объекты для unit-тестов

Этот подход обеспечивает масштабируемость, тестируемость и поддерживаемость системы.