Skip to content

convoke.bases

Decentralized module dependency declaration and initialization

Bases provide a similar concept to Django's AppConfig, and act as a central place for each module to register important things like signal handlers and template context processors, without needing a global central object.

A single HQ acts as the coordinator for a suite of Bases. At runtime, an application instantiates an HQ, providing a list of dependencies (dotted strings, similar to Django's INSTALLED_APPS setting). Each dependency is a dotted string to a module or package containing a Base subclass named Main. Bases may also declare their own dependencies.

This system allows us to avoid module-level code dependencies that depend overly on import order and global state, and allows a better separation of initialization and execution.

convoke.bases.HQ

The HQ is the special root Base.

The HQ is directly instantiated by a client code entrypoint, rather than discovered by the dependency loader.

hq = HQ(config=MyConfig(), dependencies=['foo'])
Source code in src/convoke/bases.py 
 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
@dataclass
class HQ:
    """The HQ is the special root Base.

    The HQ is directly instantiated by a client code entrypoint, rather than
    discovered by the dependency loader.

        hq = HQ(config=MyConfig(), dependencies=['foo'])
    """

    config: BaseConfig = field(default_factory=BaseConfig, repr=False)

    bases: dict[str, Base] = field(init=False, default_factory=dict, repr=False)
    signal_receivers: dict[Type[Signal], set[Receiver]] = field(init=False, default_factory=lambda: defaultdict(set))
    mountpoints: MountpointDict[Type[Mountpoint], Mountpoint] = field(init=False, default_factory=MountpointDict)

    hq: HQ = field(init=False)

    current_instance: ClassVar[ContextVar] = ContextVar("current_instance")

    def __post_init__(self):
        self.hq = self
        self.current_instance.set(self)

    @classmethod
    def get_current(cls):
        """Return the instance of HQ for the current context."""
        return cls.current_instance.get()

    def reset(self):
        """Reset this HQ and its associated Bases.

        Primarily, this re-establishes this instance as the current HQ
        instance, and re-initializes bases.

        """
        self.current_instance.set(self)
        for base in self.bases.values():
            base.reset()

    def load_dependencies(self, dependencies: Sequence[str]):
        """Load peripheral Base dependencies.

        :param Sequence[str] dependencies: a list of dotted paths to
            modules/packages that contain a Base subclass named `Main`.
        """
        load_dependencies(self, dependencies)
        for base in self.bases.values():
            base.ready()
            logging.debug(f"{base.__module__} reports ready")

    def connect_signal_receiver(self, signal_class: Type[Signal], receiver: Receiver):
        """Connect a receiver function to the given Signal subclass.

        All connections are local to this HQ instance. Mostly used
        internally via the `Base.responds(SignalSubclass)` decorator.

        :param Type[Signal] signal_class: The Signal subclass to connect to.
        :param Receiver receiver: a Callable that accepts a message of the type
            defined on the Signal subclass.

        """
        self.signal_receivers[signal_class].add(receiver)

    def disconnect_signal_receiver(self, signal_class: Type[Signal], receiver: Receiver):
        """Disconnect a receiver function previously connected to the given Signal subclass.

        :param Type[Signal] signal_class: The Signal subclass to disconnect from.
        :param Receiver receiver: a previously-connected Callable.
        """
        self.signal_receivers[signal_class].discard(receiver)

    async def send_signal(self, signal_class: Type[Signal], msg):
        """Send a Message to all receivers of the given Signal subclass.

        :param Type[Signal] signal_class: The Signal subclass to send
        :param Any msg: An instance of signal_class.Message
        """
        for receiver in self.signal_receivers[signal_class]:
            try:
                if is_async_callable(receiver):
                    await receiver(msg)
                else:
                    receiver(msg)
            except Exception:  # pragma: nocover
                # It's important that we swallow the exception, log
                # it, and soldier on. We don't need to cover this
                # branch though.
                logging.exception(
                    f"Exception occurred while sending {signal_class!r}:\nReceiver {receiver!r}\n Message: {msg!r}"
                )

connect_signal_receiver(signal_class, receiver)

Connect a receiver function to the given Signal subclass.

All connections are local to this HQ instance. Mostly used internally via the Base.responds(SignalSubclass) decorator.

Parameters:

Name Type Description Default
signal_class Type[Signal]

The Signal subclass to connect to.

 required
receiver Receiver

a Callable that accepts a message of the type defined on the Signal subclass.

 required
Source code in src/convoke/bases.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def connect_signal_receiver(self, signal_class: Type[Signal], receiver: Receiver):
    """Connect a receiver function to the given Signal subclass.

    All connections are local to this HQ instance. Mostly used
    internally via the `Base.responds(SignalSubclass)` decorator.

    :param Type[Signal] signal_class: The Signal subclass to connect to.
    :param Receiver receiver: a Callable that accepts a message of the type
        defined on the Signal subclass.

    """
    self.signal_receivers[signal_class].add(receiver)

disconnect_signal_receiver(signal_class, receiver)

Disconnect a receiver function previously connected to the given Signal subclass.

Parameters:

Name Type Description Default
signal_class Type[Signal]

The Signal subclass to disconnect from.

 required
receiver Receiver

a previously-connected Callable.

 required
Source code in src/convoke/bases.py 
103
104
105
106
107
108
109
def disconnect_signal_receiver(self, signal_class: Type[Signal], receiver: Receiver):
    """Disconnect a receiver function previously connected to the given Signal subclass.

    :param Type[Signal] signal_class: The Signal subclass to disconnect from.
    :param Receiver receiver: a previously-connected Callable.
    """
    self.signal_receivers[signal_class].discard(receiver)

get_current() classmethod

Return the instance of HQ for the current context.

Source code in src/convoke/bases.py 
63
64
65
66
@classmethod
def get_current(cls):
    """Return the instance of HQ for the current context."""
    return cls.current_instance.get()

load_dependencies(dependencies)

Load peripheral Base dependencies.

Parameters:

Name Type Description Default
dependencies Sequence[str]

a list of dotted paths to modules/packages that contain a Base subclass named Main.

 required
Source code in src/convoke/bases.py 
79
80
81
82
83
84
85
86
87
88
def load_dependencies(self, dependencies: Sequence[str]):
    """Load peripheral Base dependencies.

    :param Sequence[str] dependencies: a list of dotted paths to
        modules/packages that contain a Base subclass named `Main`.
    """
    load_dependencies(self, dependencies)
    for base in self.bases.values():
        base.ready()
        logging.debug(f"{base.__module__} reports ready")

reset()

Reset this HQ and its associated Bases.

Primarily, this re-establishes this instance as the current HQ instance, and re-initializes bases.

Source code in src/convoke/bases.py 
68
69
70
71
72
73
74
75
76
77
def reset(self):
    """Reset this HQ and its associated Bases.

    Primarily, this re-establishes this instance as the current HQ
    instance, and re-initializes bases.

    """
    self.current_instance.set(self)
    for base in self.bases.values():
        base.reset()

send_signal(signal_class, msg) async

Send a Message to all receivers of the given Signal subclass.

Parameters:

Name Type Description Default
signal_class Type[Signal]

The Signal subclass to send

 required
msg Any

An instance of signal_class.Message

 required
Source code in src/convoke/bases.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
async def send_signal(self, signal_class: Type[Signal], msg):
    """Send a Message to all receivers of the given Signal subclass.

    :param Type[Signal] signal_class: The Signal subclass to send
    :param Any msg: An instance of signal_class.Message
    """
    for receiver in self.signal_receivers[signal_class]:
        try:
            if is_async_callable(receiver):
                await receiver(msg)
            else:
                receiver(msg)
        except Exception:  # pragma: nocover
            # It's important that we swallow the exception, log
            # it, and soldier on. We don't need to cover this
            # branch though.
            logging.exception(
                f"Exception occurred while sending {signal_class!r}:\nReceiver {receiver!r}\n Message: {msg!r}"
            )

convoke.bases.Base

A Base organizes an app within a Convoke project.

For base discovery, an app should provide a subclass named Main in the app's primary module:

class Main(Base):
    config_class: MyConfig = MyConfig

    # Dependencies are dotted paths to modules containing a
    # Main subclass:
    dependencies = ['foo', 'foo.bar']

Base is similar in concept to Django's AppConfig.

Source code in src/convoke/bases.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
class Base(metaclass=BaseMeta):
    """A Base organizes an app within a Convoke project.

    For base discovery, an app should provide a subclass named `Main`
    in the app's primary module:

        class Main(Base):
            config_class: MyConfig = MyConfig

            # Dependencies are dotted paths to modules containing a
            # Main subclass:
            dependencies = ['foo', 'foo.bar']

    Base is similar in concept to Django's `AppConfig`.
    """

    hq: HQ

    bases: dict[str, Base] = field(init=False, default_factory=dict, repr=False)

    config: BaseConfig = field(init=False, repr=False)

    dependencies: ClassVar[Sequence[str]] = ()
    config_class: ClassVar[Type[BaseConfig]] = BaseConfig
    current_instance: ClassVar[ContextVar]

    def __init_subclass__(cls):
        cls.current_instance = ContextVar("current_instance")

    def __post_init__(self):
        self.reset()

    def ready(self):
        """Make the base ready for action."""
        self.on_ready()

    field = field
    responds = responds

    def reset(self):
        """Reset the base, reloading configuration and initialization."""
        self.config = self.config_class.from_config(self.hq.config)
        self.on_init()
        self._register_special_methods()
        self.current_instance.set(self)

    def on_init(self):
        """Subclass-overridable method to call at the end of initialization"""
        pass

    def on_ready(self):
        """Subclass-overridable method to call after all Bases ready"""
        pass

    @classmethod
    def get_current(cls):
        """Return the current instance of this Base for the current context."""
        return cls.current_instance.get()

    def _register_special_methods(self):
        """Look for and register specially-decorated Base methods."""
        # We need to inspect members of the class, not the instance,
        # to avoid tripping over any descriptors before we're ready.
        for name, func in inspect.getmembers(self.__class__, inspect.isfunction):
            if inspect.ismethod(method := getattr(self, name)):
                # Register signals
                signals = getattr(func, "__signals__", ())
                for signal in signals:
                    self.hq.connect_signal_receiver(signal, method)

                # Register mountpoints
                mountpoints = getattr(method.__func__, "__mountpoints__", ())
                for mountpoint in mountpoints:
                    self.hq.mountpoints[mountpoint].mount(method)
            else:  # pragma: nocover
                pass

get_current() classmethod

Return the current instance of this Base for the current context.

Source code in src/convoke/bases.py 
233
234
235
236
@classmethod
def get_current(cls):
    """Return the current instance of this Base for the current context."""
    return cls.current_instance.get()

on_init()

Subclass-overridable method to call at the end of initialization

Source code in src/convoke/bases.py 
225
226
227
def on_init(self):
    """Subclass-overridable method to call at the end of initialization"""
    pass

on_ready()

Subclass-overridable method to call after all Bases ready

Source code in src/convoke/bases.py 
229
230
231
def on_ready(self):
    """Subclass-overridable method to call after all Bases ready"""
    pass

ready()

Make the base ready for action.

Source code in src/convoke/bases.py 
211
212
213
def ready(self):
    """Make the base ready for action."""
    self.on_ready()

reset()

Reset the base, reloading configuration and initialization.

Source code in src/convoke/bases.py 
218
219
220
221
222
223
def reset(self):
    """Reset the base, reloading configuration and initialization."""
    self.config = self.config_class.from_config(self.hq.config)
    self.on_init()
    self._register_special_methods()
    self.current_instance.set(self)