ArchiveBox/archivebox/plugantic/base_binary.py
Nick Sweeting bb65b2dbec
Some checks are pending
CodeQL / Analyze (python) (push) Waiting to run
Build Debian package / build (push) Waiting to run
Build Docker image / buildx (push) Waiting to run
Build Homebrew package / build (push) Waiting to run
Build GitHub Pages website / build (push) Waiting to run
Build GitHub Pages website / deploy (push) Blocked by required conditions
Run linters / lint (push) Waiting to run
Build Pip package / build (push) Waiting to run
Run tests / python_tests (ubuntu-22.04, 3.11) (push) Waiting to run
Run tests / docker_tests (push) Waiting to run
move almost all config into new archivebox.CONSTANTS
2024-09-25 05:10:09 -07:00

106 lines
3.6 KiB
Python

__package__ = "archivebox.plugantic"
from typing import Dict, List
from typing_extensions import Self
from benedict import benedict
from pydantic import Field, InstanceOf, validate_call
from pydantic_pkgr import (
Binary,
BinProvider,
BinProviderName,
ProviderLookupDict,
AptProvider,
BrewProvider,
EnvProvider,
)
from django.conf import settings
import archivebox
from .base_hook import BaseHook, HookType
class BaseBinProvider(BaseHook, BinProvider):
hook_type: HookType = "BINPROVIDER"
# def on_get_abspath(self, bin_name: BinName, **context) -> Optional[HostBinPath]:
# Class = super()
# get_abspath_func = lambda: Class.on_get_abspath(bin_name, **context)
# # return cache.get_or_set(f'bin:abspath:{bin_name}', get_abspath_func)
# return get_abspath_func()
# def on_get_version(self, bin_name: BinName, abspath: Optional[HostBinPath]=None, **context) -> SemVer | None:
# Class = super()
# get_version_func = lambda: Class.on_get_version(bin_name, abspath, **context)
# # return cache.get_or_set(f'bin:version:{bin_name}:{abspath}', get_version_func)
# return get_version_func()
def register(self, settings, parent_plugin=None):
# self._plugin = parent_plugin # for debugging only, never rely on this!
settings.BINPROVIDERS = getattr(settings, "BINPROVIDERS", None) or benedict({})
settings.BINPROVIDERS[self.id] = self
super().register(settings, parent_plugin=parent_plugin)
@property
def admin_url(self) -> str:
# e.g. /admin/environment/binproviders/NpmBinProvider/ TODO
return "/admin/environment/binaries/"
class BaseBinary(BaseHook, Binary):
hook_type: HookType = "BINARY"
binproviders_supported: List[InstanceOf[BinProvider]] = Field(default_factory=list, alias="binproviders")
provider_overrides: Dict[BinProviderName, ProviderLookupDict] = Field(default_factory=dict, alias="overrides")
def register(self, settings, parent_plugin=None):
# self._plugin = parent_plugin # for debugging only, never rely on this!
settings.BINARIES = getattr(settings, "BINARIES", None) or benedict({})
settings.BINARIES[self.id] = self
super().register(settings, parent_plugin=parent_plugin)
@staticmethod
def symlink_to_lib(binary, bin_dir=None) -> None:
bin_dir = bin_dir or archivebox.CONSTANTS.LIB_BIN_DIR
if not (binary.abspath and binary.abspath.exists()):
return
bin_dir.mkdir(parents=True, exist_ok=True)
symlink = bin_dir / binary.name
symlink.unlink(missing_ok=True)
symlink.symlink_to(binary.abspath)
@validate_call
def load(self, **kwargs) -> Self:
binary = super().load(**kwargs)
self.symlink_to_lib(binary=binary, bin_dir=archivebox.CONSTANTS.LIB_BIN_DIR)
return binary
@validate_call
def install(self, **kwargs) -> Self:
binary = super().install(**kwargs)
self.symlink_to_lib(binary=binary, bin_dir=archivebox.CONSTANTS.LIB_BIN_DIR)
return binary
@validate_call
def load_or_install(self, **kwargs) -> Self:
binary = super().load_or_install(**kwargs)
self.symlink_to_lib(binary=binary, bin_dir=archivebox.CONSTANTS.LIB_BIN_DIR)
return binary
@property
def admin_url(self) -> str:
# e.g. /admin/environment/config/LdapConfig/
return f"/admin/environment/binaries/{self.name}/"
apt = AptProvider()
brew = BrewProvider()
env = EnvProvider()