wehub-resource-sync
57 lines
1.4 KiB
Python
57 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
from abc import ABC
|
|
from typing import Literal
|
|
|
|
from . import utils
|
|
|
|
EventTypes = Literal["plugin_registered",]
|
|
|
|
|
|
class Plugin(ABC): # noqa: B024
|
|
registered_plugins: list[Plugin] = []
|
|
emitter: utils.EventEmitter[EventTypes] = utils.EventEmitter()
|
|
|
|
# TODO(theomonnom): make logger mandatory once all plugins have been updated
|
|
def __init__(
|
|
self,
|
|
title: str,
|
|
version: str,
|
|
package: str,
|
|
logger: logging.Logger | None = None,
|
|
) -> None:
|
|
self._title = title
|
|
self._version = version
|
|
self._package = package
|
|
self._logger = logger
|
|
|
|
@classmethod
|
|
def register_plugin(cls, plugin: Plugin) -> None:
|
|
if threading.current_thread() != threading.main_thread():
|
|
raise RuntimeError("Plugins must be registered on the main thread")
|
|
|
|
cls.registered_plugins.append(plugin)
|
|
cls.emitter.emit("plugin_registered", plugin)
|
|
|
|
# plugin can implement an optional download_files method
|
|
def download_files(self) -> None: # noqa: B027
|
|
pass
|
|
|
|
@property
|
|
def package(self) -> str:
|
|
return self._package
|
|
|
|
@property
|
|
def title(self) -> str:
|
|
return self._title
|
|
|
|
@property
|
|
def version(self) -> str:
|
|
return self._version
|
|
|
|
@property
|
|
def logger(self) -> logging.Logger | None:
|
|
return self._logger
|