mirror of
https://github.com/ArchiveBox/ArchiveBox.git
synced 2026-04-05 23:37:58 +10:00
cleanup plugantic and pkg apps, make BaseHook actually create its own settings
This commit is contained in:
Nick Sweeting
@@ -5,9 +5,8 @@ import inspect
|
||||
from pathlib import Path
|
||||
|
||||
from django.apps import AppConfig
|
||||
from django.core.checks import register
|
||||
|
||||
from typing import List, ClassVar, Type, Dict
|
||||
from typing import List, Type, Dict
|
||||
from typing_extensions import Self
|
||||
|
||||
from pydantic import (
|
||||
@@ -20,142 +19,99 @@ from pydantic import (
|
||||
validate_call,
|
||||
)
|
||||
|
||||
from .base_configset import BaseConfigSet
|
||||
from .base_binary import BaseBinProvider, BaseBinary
|
||||
from .base_extractor import BaseExtractor
|
||||
from .base_replayer import BaseReplayer
|
||||
from .base_check import BaseCheck
|
||||
from .base_admindataview import BaseAdminDataView
|
||||
from .base_hook import BaseHook, HookType
|
||||
|
||||
from ..config import ANSI, AttrDict
|
||||
from ..config import AttrDict
|
||||
|
||||
|
||||
class BasePlugin(BaseModel):
|
||||
model_config = ConfigDict(arbitrary_types_allowed=True, extra='ignore', populate_by_name=True)
|
||||
|
||||
# Required by AppConfig:
|
||||
name: str = Field() # e.g. 'builtin_plugins.singlefile' (DottedImportPath)
|
||||
app_label: str = Field() # e.g. 'singlefile' (one-word machine-readable representation, to use as url-safe id/db-table prefix_/attr name)
|
||||
verbose_name: str = Field() # e.g. 'SingleFile' (human-readable *short* label, for use in column names, form labels, etc.)
|
||||
verbose_name: str = Field() # e.g. 'SingleFile' (human-readable *short* label, for use in column names, form labels, etc.)
|
||||
|
||||
# All the hooks the plugin will install:
|
||||
configs: List[InstanceOf[BaseConfigSet]] = Field(default=[])
|
||||
binproviders: List[InstanceOf[BaseBinProvider]] = Field(default=[]) # e.g. [Binary(name='yt-dlp')]
|
||||
binaries: List[InstanceOf[BaseBinary]] = Field(default=[]) # e.g. [Binary(name='yt-dlp')]
|
||||
extractors: List[InstanceOf[BaseExtractor]] = Field(default=[])
|
||||
replayers: List[InstanceOf[BaseReplayer]] = Field(default=[])
|
||||
checks: List[InstanceOf[BaseCheck]] = Field(default=[])
|
||||
admindataviews: List[InstanceOf[BaseAdminDataView]] = Field(default=[])
|
||||
hooks: List[InstanceOf[BaseHook]] = Field(default=[])
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def id(self) -> str:
|
||||
return self.__class__.__name__
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def plugin_module(self) -> str: # DottedImportPath
|
||||
""" "
|
||||
Dotted import path of the plugin's module (after its loaded via settings.INSTALLED_APPS).
|
||||
e.g. 'archivebox.builtin_plugins.npm.apps.NpmPlugin' -> 'builtin_plugins.npm'
|
||||
"""
|
||||
return f"{self.__module__}.{self.__class__.__name__}".split("archivebox.", 1)[-1].rsplit('.apps.', 1)[0]
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def plugin_dir(self) -> Path:
|
||||
return Path(inspect.getfile(self.__class__)).parent.resolve()
|
||||
|
||||
@model_validator(mode='after')
|
||||
def validate(self) -> Self:
|
||||
"""Validate the plugin's build-time configuration here before it's registered in Django at runtime."""
|
||||
|
||||
assert self.name and self.app_label and self.verbose_name, f'{self.__class__.__name__} is missing .name or .app_label or .verbose_name'
|
||||
assert self.app_label and self.app_label and self.verbose_name, f'{self.__class__.__name__} is missing .name or .app_label or .verbose_name'
|
||||
|
||||
assert json.dumps(self.model_json_schema(), indent=4), f'Plugin {self.name} has invalid JSON schema.'
|
||||
assert json.dumps(self.model_json_schema(), indent=4), f"Plugin {self.plugin_module} has invalid JSON schema."
|
||||
return self
|
||||
|
||||
@property
|
||||
def AppConfig(plugin_self) -> Type[AppConfig]:
|
||||
"""Generate a Django AppConfig class for this plugin."""
|
||||
|
||||
|
||||
class PluginAppConfig(AppConfig):
|
||||
"""Django AppConfig for plugin, allows it to be loaded as a Django app listed in settings.INSTALLED_APPS."""
|
||||
name = plugin_self.name
|
||||
name = plugin_self.plugin_module
|
||||
app_label = plugin_self.app_label
|
||||
verbose_name = plugin_self.verbose_name
|
||||
|
||||
default_auto_field = 'django.db.models.AutoField'
|
||||
|
||||
|
||||
def ready(self):
|
||||
from django.conf import settings
|
||||
|
||||
# plugin_self.validate()
|
||||
plugin_self.register(settings)
|
||||
|
||||
return PluginAppConfig
|
||||
|
||||
@computed_field
|
||||
|
||||
@property
|
||||
def BINPROVIDERS(self) -> Dict[str, BaseBinProvider]:
|
||||
return AttrDict({binprovider.name: binprovider for binprovider in self.binproviders})
|
||||
|
||||
@computed_field
|
||||
def HOOKS_BY_ID(self) -> Dict[str, InstanceOf[BaseHook]]:
|
||||
return AttrDict({hook.id: hook for hook in self.hooks})
|
||||
|
||||
@property
|
||||
def BINARIES(self) -> Dict[str, BaseBinary]:
|
||||
return AttrDict({binary.python_name: binary for binary in self.binaries})
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def CONFIGS(self) -> Dict[str, BaseConfigSet]:
|
||||
return AttrDict({config.name: config for config in self.configs})
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def EXTRACTORS(self) -> Dict[str, BaseExtractor]:
|
||||
return AttrDict({extractor.name: extractor for extractor in self.extractors})
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def REPLAYERS(self) -> Dict[str, BaseReplayer]:
|
||||
return AttrDict({replayer.name: replayer for replayer in self.replayers})
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def CHECKS(self) -> Dict[str, BaseCheck]:
|
||||
return AttrDict({check.name: check for check in self.checks})
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def ADMINDATAVIEWS(self) -> Dict[str, BaseCheck]:
|
||||
return AttrDict({admindataview.name: admindataview for admindataview in self.admindataviews})
|
||||
def HOOKS_BY_TYPE(self) -> Dict[HookType, Dict[str, InstanceOf[BaseHook]]]:
|
||||
hooks = AttrDict({})
|
||||
for hook in self.hooks:
|
||||
hooks[hook.hook_type] = hooks.get(hook.hook_type) or AttrDict({})
|
||||
hooks[hook.hook_type][hook.id] = hook
|
||||
return hooks
|
||||
|
||||
|
||||
def register(self, settings=None):
|
||||
"""Loads this plugin's configs, binaries, extractors, and replayers into global Django settings at runtime."""
|
||||
|
||||
|
||||
if settings is None:
|
||||
from django.conf import settings as django_settings
|
||||
settings = django_settings
|
||||
|
||||
assert all(hasattr(settings, key) for key in ['PLUGINS', 'CONFIGS', 'BINARIES', 'EXTRACTORS', 'REPLAYERS', 'ADMINDATAVIEWS']), 'Tried to register plugin in settings but couldnt find required global dicts in settings.'
|
||||
assert json.dumps(self.model_json_schema(), indent=4), f'Plugin {self.plugin_module} has invalid JSON schema.'
|
||||
|
||||
assert json.dumps(self.model_json_schema(), indent=4), f'Plugin {self.name} has invalid JSON schema.'
|
||||
|
||||
assert self.app_label not in settings.PLUGINS, f'Tried to register plugin {self.name} but it conflicts with existing plugin of the same name ({self.app_label}).'
|
||||
assert self.id not in settings.PLUGINS, f'Tried to register plugin {self.plugin_module} but it conflicts with existing plugin of the same name ({self.app_label}).'
|
||||
|
||||
### Mutate django.conf.settings... values in-place to include plugin-provided overrides
|
||||
settings.PLUGINS[self.app_label] = self
|
||||
settings.PLUGINS[self.id] = self
|
||||
|
||||
for config in self.CONFIGS.values():
|
||||
config.register(settings, parent_plugin=self)
|
||||
|
||||
for binprovider in self.BINPROVIDERS.values():
|
||||
binprovider.register(settings, parent_plugin=self)
|
||||
|
||||
for binary in self.BINARIES.values():
|
||||
binary.register(settings, parent_plugin=self)
|
||||
|
||||
for extractor in self.EXTRACTORS.values():
|
||||
extractor.register(settings, parent_plugin=self)
|
||||
for hook in self.hooks:
|
||||
hook.register(settings, parent_plugin=self)
|
||||
|
||||
for replayer in self.REPLAYERS.values():
|
||||
replayer.register(settings, parent_plugin=self)
|
||||
|
||||
for check in self.CHECKS.values():
|
||||
check.register(settings, parent_plugin=self)
|
||||
|
||||
for admindataview in self.ADMINDATAVIEWS.values():
|
||||
admindataview.register(settings, parent_plugin=self)
|
||||
|
||||
# TODO: add parsers? custom templates? persona fixtures?
|
||||
|
||||
plugin_prefix, plugin_shortname = self.name.split('.', 1)
|
||||
|
||||
print(
|
||||
f' > {ANSI.black}{plugin_prefix.upper().replace("_PLUGINS", "").ljust(15)} ' +
|
||||
f'{ANSI.lightyellow}{plugin_shortname.ljust(12)} ' +
|
||||
f'{ANSI.black}CONFIGSx{len(self.configs)} BINARIESx{len(self.binaries)} EXTRACTORSx{len(self.extractors)} REPLAYERSx{len(self.replayers)} CHECKSx{len(self.CHECKS)} ADMINDATAVIEWSx{len(self.ADMINDATAVIEWS)}{ANSI.reset}'
|
||||
)
|
||||
print('√ REGISTERED PLUGIN:', self.plugin_module)
|
||||
|
||||
# @validate_call
|
||||
# def install_binaries(self) -> Self:
|
||||
@@ -169,7 +125,7 @@ class BasePlugin(BaseModel):
|
||||
@validate_call
|
||||
def load_binaries(self, cache=True) -> Self:
|
||||
new_binaries = []
|
||||
for idx, binary in enumerate(self.binaries):
|
||||
for idx, binary in enumerate(self.HOOKS_BY_TYPE['BINARY'].values()):
|
||||
new_binaries.append(binary.load(cache=cache) or binary)
|
||||
return self.model_copy(update={
|
||||
'binaries': new_binaries,
|
||||
@@ -184,20 +140,6 @@ class BasePlugin(BaseModel):
|
||||
# 'binaries': new_binaries,
|
||||
# })
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def module_dir(self) -> Path:
|
||||
return Path(inspect.getfile(self.__class__)).parent.resolve()
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def module_path(self) -> str: # DottedImportPath
|
||||
""""
|
||||
Dotted import path of the plugin's module (after its loaded via settings.INSTALLED_APPS).
|
||||
e.g. 'archivebox.builtin_plugins.npm'
|
||||
"""
|
||||
return self.name.strip('archivebox.')
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user