Reestructuración de proyecto SaaS B2B · Aplicando estructura-proyectos-python
datapulse/core.py tiene 1.240 líneas mezclando modelos de datos, lógica de negocio, inferencia ML y rutas HTTP. Imposible testear en aislamiento, conflictos de merge frecuentes (4 ingenieros editan el mismo archivo), cobertura de tests <30%.
HTTP handlers, request/response, autenticación JWT
Lógica de negocio, orquestación, reglas de dominio
Acceso a datos, queries SQLAlchemy async, caché Redis
Entidades de dominio (SQLAlchemy ORM), Pydantic schemas
Configuración, excepciones comunes, utilidades transversales
"""DataPulse Services — Lógica de negocio del dominio.""" from .user_service import UserService from .campaign_service import CampaignService from .prediction_service import PredictionService from .exceptions import ServiceError, ValidationError, NotFoundError __all__ = [ "UserService", "CampaignService", "PredictionService", "ServiceError", "ValidationError", "NotFoundError", ] # Helpers internos — NO exportados, sin listar en __all__ # from ._internal_validators import _validate_campaign_dates
"""Servicio de predicciones ML — orquesta modelos churn y LTV.""" from __future__ import annotations from datapulse.models import User, Campaign from datapulse.repositories import UserRepository, CampaignRepository from datapulse.ml import ChurnModel, LTVModel from datapulse.schemas import PredictionResult from .exceptions import NotFoundError, ServiceError import logging logger = logging.getLogger(__name__) class PredictionService: """Orquesta predicciones de churn y LTV para usuarios DataPulse.""" def __init__( self, user_repo: UserRepository, campaign_repo: CampaignRepository, churn_model: ChurnModel, ltv_model: LTVModel, ) -> None: self._user_repo = user_repo self._campaign_repo = campaign_repo self._churn_model = churn_model self._ltv_model = ltv_model async def predict_churn(self, user_id: int) -> PredictionResult: """Predice probabilidad de churn para un usuario específico.""" user = await self._user_repo.get_by_id(user_id) if not user: raise NotFoundError(f"Usuario {user_id} no encontrado") features = await self._extract_churn_features(user) score = self._churn_model.predict_proba(features) logger.info("Churn score para user=%d: %.3f", user_id, score) return PredictionResult(user_id=user_id, churn_score=score) async def predict_ltv(self, user_id: int, months: int = 12) -> PredictionResult: """Predice LTV estimado a N meses.""" user = await self._user_repo.get_by_id(user_id) if not user: raise NotFoundError(f"Usuario {user_id} no encontrado") features = await self._extract_ltv_features(user, months) ltv_eur = self._ltv_model.predict(features) return PredictionResult(user_id=user_id, ltv_eur=ltv_eur, horizon_months=months) async def _extract_churn_features(self, user: User) -> dict: """Helper privado — no forma parte de la API pública.""" campaigns = await self._campaign_repo.get_recent_by_user(user.id, days=30) return { "days_since_last_login": user.days_since_last_login, "campaigns_last_30d": len(campaigns), "plan_tier": user.plan.value, "mrr": float(user.mrr), }
# pyproject.toml — reemplaza setup.py/setup.cfg [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [project] name = "datapulse" version = "2.0.0" requires-python = ">=3.11" dependencies = [ "fastapi>=0.110", "sqlalchemy[asyncio]>=2.0", "pydantic>=2.0", "scikit-learn>=1.4", "redis[hiredis]>=5.0", ] [tool.hatch.build.targets.wheel] packages = ["src/datapulse"] [tool.pytest.ini_options] testpaths = ["tests"] asyncio_mode = "auto" [tool.mypy] # src/ layout — mypy encuentra el paquete aquí mypy_path = "src" strict = true [tool.ruff.lint] select = ["E", "F", "I", "UP"] # I = isort (imports ordenados), UP = pyupgrade
core.py (1.240 líneas) dividido en 8 archivos de servicio + 3 de modelos + 4 de API. Máximo ~180 líneas por archivo.
✅ AplicadoCada paquete (services/, models/, repositories/) define __all__ en su __init__.py. Los helpers privados usan prefijo _ y no se exportan.
✅ AplicadoMáximo 4 niveles de profundidad: src/datapulse/services/prediction_service.py. Sin anidamiento extra innecesario.
✅ Aplicadotests/services/test_prediction_service.py espeja src/datapulse/services/. Permite pytest con cobertura aislada por capa.
✅ Aplicadosrc/datapulse/__init__.py exporta la API pública: from datapulse import PredictionService, UserService — una línea de import.
✅ Aplicado4 capas: api → services → repositories → models. Dependencias en una sola dirección. Sin ciclos. Config y ml como capas ortogonales.
✅ AplicadoLa capa ml/ usa estructura de dominio: churn_model.py, ltv_model.py separados. Integrations/ agrupa conectores externos como sub-dominio.
✅ Aplicado (parcial)"""DataPulse Analytics SDK — imports de nivel raíz.""" from .services import UserService, CampaignService, PredictionService from .models import User, Campaign, Prediction from .schemas import PredictionResult, CampaignMetrics, UserProfile from .config import Settings from .exceptions import DataPulseError, NotFoundError __version__ = "2.0.0" __all__ = [ # Services "UserService", "CampaignService", "PredictionService", # Models "User", "Campaign", "Prediction", # Schemas "PredictionResult", "CampaignMetrics", "UserProfile", # Config & Errors "Settings", "DataPulseError", "NotFoundError", ] # Uso externo — SDK de DataPulse: # from datapulse import PredictionService, Settings # service = PredictionService(settings=Settings()) # result = await service.predict_churn(user_id=42)