Rework

backend/nango/cogs/__init__.py

@@ -0,0 +1,10 @@
+"""Cogs are nango's plugins that give it flexibility.
+
+Every Nango function is linked to a cog.
+
+Note: Order of the __all__ matches the order of cogs execution.
+"""
+
+from .serializer_cog import SerializerCog
+
+__all__ = ["SerializerCog"]

backend/nango/cogs/serializer_cog.py

@@ -0,0 +1,15 @@
+from nango.utils import AbstractCog
+
+
+class SerializerCog(AbstractCog):
+    """Generate serializer from each django's models.
+
+    Settings:
+    --------
+    """
+
+    id = "serializer"
+
+    def run(self) -> None:
+        """Generate a serializer for each django's model."""
+        super().run()

backend/nango/management/commands/nango.py

@@ -0,0 +1,15 @@
+"""CLI to run the Nango's bridge."""
+
+from __future__ import annotations
+
+from django.core.management.base import BaseCommand, CommandParser
+
+
+class Command(BaseCommand):
+    """Cli to manage the bridge."""
+
+    def add_arguments(self, parser: CommandParser) -> None:
+        """."""
+
+    def handle(self, *args: list, **options: dict) -> None:
+        """Handle argument and run Nango's cogs."""

backend/nango/utils.py → backend/nango/utils/__init__.py

@@ -1,3 +1,10 @@
+from nango.utils.abstract_cog import AbstractCog
+
+__all__ = [
+    "AbstractCog",
+]
+
+
 def setup_django() -> None:
     """Setup django environment."""
     import os

backend/nango/utils/abstract_cog.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from django.db import models
+
+
+class AbstractCog(ABC):
+    """Cog's base class, that gives the same basic functionalities for all cogs.
+
+    If settings are not defined or is `{}`, the default settings will be used.
+    Otherwise, if the settings are `None`, the Cog will not run.
+    If one of the settings is not specified, the default value will be used.
+    """
+
+    id = None
+
+    def __init__(self, model: models.Model, settings: dict[str, any]) -> None:
+        """Setup the cogs.
+
+        Elements of the setup:
+
+            - Settings
+        """
+        self.model = model
+        self.settings = settings.get(self.id, {})
+
+    @abstractmethod
+    def run(self) -> None:
+        """Execute the cog logic."""
+        if not self.is_executable():
+            return
+
+    def is_executable(self) -> bool:
+        """Indicate if the cog can run or not."""
+        return self.settings is not None

backend/nango/utils/cogs_runner.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from importlib import import_module
+from typing import TYPE_CHECKING
+
+from django.apps import apps
+
+from nango import cogs
+
+if TYPE_CHECKING:
+    from django.db import models
+
+
+class CogsRunner:
+    """Load and execute cogs.
+
+    For the order execution, follow the nango.cogs.__all__ order.
+
+    Settings:
+    --------
+    Nango's settings can be defined for each cog at the model level.
+    A model can have a staticmethod called `nango` that returns a None (for no generation) or a dict (with settings for each cogs).
+    """
+
+    model_filter_keywords: list[str] = ("django", "django_celery_beat", "rest_framework")
+
+    def get_cogs_classes(self) -> list[Callable]:
+        """Return cogs's classes to execute."""
+        return [getattr(import_module("nango.cogs"), cog_str) for cog_str in cogs.__all__]
+
+    def get_settings_from_model(self, model: models.Model) -> dict[str, any]:
+        """Return settings define in a model."""
+        model_nango_method: Callable | None = getattr(model, "nango", None)
+        if not isinstance(model_nango_method, Callable):
+            if model_nango_method is None:
+                # No settings defined in the model
+                return {}
+            error_msg: str = f"{model}.nango must be a staticmethod, not a property."
+            raise TypeError(error_msg)
+        return model_nango_method()
+
+    def _is_filtered_model(self, model: models.Model) -> bool:
+        """Indicate if a model is filter or not.
+
+        If the model is filter, we don't run it.
+        """
+        return any(keyword in str(model) for keyword in self.model_filter_keywords)
+
+    def run_cogs(self) -> None:
+        """Run all existing cogs."""
+        cogs_classes = self.get_cogs_classes()
+
+        for model in apps.get_models():
+            if self._is_filtered_model(model):
+                continue
+            for cog in cogs_classes:
+                settings = self.get_settings_from_model(model)
+                if settings is None:
+                    # This model doesn't want any cogs.
+                    continue
+
+                # Run the cogs
+                instantiated_cog = cog(model=model, settings=settings)
+                instantiated_cog.run()
+
+
+if __name__ == "django.core.management.commands.shell":
+    CogsRunner().run_cogs()