Skip to content

Reference

django_modulith.interface_registry

InterfaceRegistry

Registry that dynamically adds interfaces as actual methods

Source code in src/django_modulith/interface_registry.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class InterfaceRegistry:
    """Registry that dynamically adds interfaces as actual methods"""

    _registered_interfaces: Set[str] = (
        set()
    )  # Track registered names to prevent overrides

    @classmethod
    def register(cls, func: Callable, name: str):
        """Register a function dynamically as a method on the class"""
        if name in cls._registered_interfaces:
            raise ValueError(
                f"interface '{name}' is already registered. Choose a unique name."
            )

        setattr(cls, name, classmethod(func))
        cls._registered_interfaces.add(name)

    @classmethod
    def list_interfaces(cls) -> Set[str]:
        """List all registered interfaces"""
        return cls._registered_interfaces

list_interfaces() classmethod

List all registered interfaces

Source code in src/django_modulith/interface_registry.py 
22
23
24
25
@classmethod
def list_interfaces(cls) -> Set[str]:
    """List all registered interfaces"""
    return cls._registered_interfaces

register(func, name) classmethod

Register a function dynamically as a method on the class

Source code in src/django_modulith/interface_registry.py 
11
12
13
14
15
16
17
18
19
20
@classmethod
def register(cls, func: Callable, name: str):
    """Register a function dynamically as a method on the class"""
    if name in cls._registered_interfaces:
        raise ValueError(
            f"interface '{name}' is already registered. Choose a unique name."
        )

    setattr(cls, name, classmethod(func))
    cls._registered_interfaces.add(name)

django_modulith.interface_decorator

interface(name=None)

Decorator that can be used with or without parameters

Usage

@interface def my_function(): ...

@interface() def my_function(): ...

@interface("custom_name") def my_function(): ...

Source code in src/django_modulith/interface_decorator.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def interface(
    name: Optional[Union[str, T]] = None,
) -> Union[Callable[[T], T], T]:
    """Decorator that can be used with or without parameters

    Usage:
        @interface
        def my_function(): ...

        @interface()
        def my_function(): ...

        @interface("custom_name")
        def my_function(): ...
    """
    func_or_name = name

    def decorator(func: T) -> T:
        service_name = (
            func.__name__
            if isinstance(func_or_name, (type(None), Callable))
            else func_or_name
        )
        InterfaceRegistry.register(func, str(service_name))

        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            return func(*args, **kwargs)

        return cast(T, wrapper)

    # Handle both @interface and @interface() cases
    if isinstance(func_or_name, Callable):
        return decorator(func_or_name)

    # Handle @interface("name") case
    return decorator

django_modulith.management.commands.generatestubs

Command

Bases: BaseCommand

Source code in src/django_modulith/management/commands/generatestubs.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class Command(BaseCommand):
    help = (
        "Generate interface stubs by scanning all installed apps for modulith.py files"
    )

    def handle(self, *args, **options):
        # Find and import all modulith.py files
        self.stdout.write("Scanning installed apps for modulith.py files...")
        modules_found = self._find_and_import_modulith_files()

        if not modules_found:
            self.stdout.write(
                self.style.WARNING("No modulith.py files found in installed apps.")
            )
        else:
            self.stdout.write(f"Found {len(modules_found)} modulith.py files.")

        # Generate stubs
        self.stdout.write("Generating interface stubs...")
        # Default to the package directory
        package_dir = Path(__file__).parent.parent.parent
        output_path = package_dir / "interface_registry.pyi"

        stub_content = self._generate_stubs()

        # Ensure the directory exists
        output_path.parent.mkdir(parents=True, exist_ok=True)

        # Write the stub file
        output_path.write_text(stub_content)

        self.stdout.write(
            self.style.SUCCESS(f"Interface stubs generated at {output_path}")
        )

    def _find_and_import_modulith_files(self) -> List[str]:
        """Find and import all modulith.py files in installed apps."""
        modules_found = []

        for app_config in apps.get_app_configs():
            app_path = Path(app_config.path)
            modulith_file = app_path / "modulith.py"

            if modulith_file.exists():
                module_name = f"{app_config.name}.modulith"
                self.stdout.write(f"Found {module_name} at {modulith_file}")

                try:
                    importlib.import_module(module_name)
                    modules_found.append(module_name)
                except Exception as e:
                    self.stdout.write(
                        self.style.ERROR(f"Error importing {module_name}: {e}")
                    )
                    raise CommandError(f"Error importing {module_name}: {e}")

        return modules_found

    def _clean_signature(self, signature):
        """Clean up the signature string by removing ~ from type annotations"""
        return str(signature).replace("~", "")

    def _extract_typevars(self, signatures):
        """Extract TypeVar definitions from signatures"""
        typevar_pattern = r"([A-Z]_co|[A-Z]_contra|[A-Z])"
        typevars = set()

        for sig in signatures:
            # Look for capital letter followed by _co, _contra or just capital letter alone
            matches = re.findall(typevar_pattern, str(sig))
            typevars.update(matches)

        # Generate TypeVar definitions
        typevar_defs = []
        for tv in sorted(typevars):
            if tv.endswith("_co"):
                base = tv[:-3]
                typevar_defs.append(f"{tv} = TypeVar('{base}', covariant=True)")
            elif tv.endswith("_contra"):
                base = tv[:-7]
                typevar_defs.append(f"{tv} = TypeVar('{base}', contravariant=True)")
            else:
                typevar_defs.append(f"{tv} = TypeVar('{tv}')")

        return typevar_defs

    def _generate_stubs(self) -> str:
        """Generate stub file content for InterfaceRegistry"""
        # Collect all signatures first to extract TypeVars
        signatures = []

        # Get signatures from built-in methods
        for _, method in inspect.getmembers(
            InterfaceRegistry, predicate=inspect.ismethod
        ):
            signatures.append(inspect.signature(method))

        # Get signatures from registered methods
        for name in InterfaceRegistry.list_interfaces():
            method = getattr(InterfaceRegistry, name)
            signatures.append(inspect.signature(method))

        # Extract TypeVars from signatures
        typevar_defs = self._extract_typevars(signatures)

        # Start building stubs
        stubs = [
            "from typing import Any, Callable, Set, ClassVar, List, TypeVar\n",
        ]

        # Add TypeVar definitions if found
        if typevar_defs:
            stubs.append("\n")
            for tv_def in typevar_defs:
                stubs.append(f"{tv_def}\n")

        stubs.extend(
            [
                "\n",
                "class InterfaceRegistry:\n",
                "    _registered_interfaces: ClassVar[Set[str]]\n",
                "\n",
            ]
        )

        # Add built-in class methods first
        for name, method in inspect.getmembers(
            InterfaceRegistry, predicate=inspect.ismethod
        ):
            if not name.startswith("_") or name == "__init__":
                signature = inspect.signature(method)
                clean_sig = self._clean_signature(signature)
                stub_line = "    @classmethod\n"
                stub_line += f"    def {name}{clean_sig}: ...\n"
                stubs.append(stub_line)

        # Add dynamically registered methods
        for name in InterfaceRegistry.list_interfaces():
            # Skip if already added (in case list_interfaces itself is registered)
            if any(f"def {name}" in line for line in stubs):
                continue

            method = getattr(InterfaceRegistry, name)
            signature = inspect.signature(method)
            clean_sig = self._clean_signature(signature)
            stub_line = "    @classmethod\n"
            stub_line += f"    def {name}{clean_sig}: ...\n"
            stubs.append(stub_line)

        return "".join(stubs)

django_modulith.management.commands.modulith

Command

Bases: Command

Source code in src/django_modulith/management/commands/modulith.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class Command(StartAppCommand):
    help = f"Add a new module to the {IMPORTLINTER_FILE} configuration"
    contract_key = "importlinter:contract:modulith_modules"

    def handle(self, *args, **options):
        super().handle(*args, **options)
        config_path = IMPORTLINTER_FILE
        config = configparser.ConfigParser()

        # Initialize or read existing config
        if os.path.exists(config_path):
            config.read(config_path)
        else:
            self._initialize_config(config)

        module_name = options["name"]
        self._add_module_to_config(config, module_name)

        # Write configuration back to file
        with open(config_path, "w") as f:
            config.write(f)

        self.stdout.write(
            self.style.SUCCESS(f"✅ {IMPORTLINTER_FILE} updated successfully!")
        )

    def _initialize_config(self, config):
        """Initialize a basic importlinter configuration"""
        config["importlinter"] = {
            "root_package": "modules",
            "include_external_packages": "n",
        }

        config[self.contract_key] = {
            "name": "Domain modules are independent",
            "type": "independence",
            "modules": "",
        }

        self.stdout.write(
            self.style.SUCCESS("Created initial importlinter configuration")
        )

    def _add_module_to_config(self, config, module_name):
        """Add a new module to the modules independence contract"""
        if self.contract_key not in config:
            self.stdout.write(
                self.style.ERROR("Modules contract section not found in config")
            )
            return

        contract_section = config[self.contract_key]
        modules = contract_section.get("modules", "")

        # Add the new module if it's not already there
        module_list = [m.strip() for m in modules.split("\n") if m.strip()]
        if module_name not in module_list:
            module_list.append(module_name)
            contract_section["modules"] = "\n".join(module_list)
            self.stdout.write(
                f"Added module '{module_name}' to importlinter configuration"
            )
        else:
            self.stdout.write(f"Module '{module_name}' already in configuration")