merge plugantic and abx, all praise be to praise our glorious pluggy gods

archivebox/abx/archivebox/base_binary.py

@@ -0,0 +1,99 @@
+__package__ = "abx.archivebox"
+
+from typing import Dict, List
+from typing_extensions import Self
+
+from pydantic import Field, InstanceOf, validate_call
+from pydantic_pkgr import (
+    Binary,
+    BinProvider,
+    BinProviderName,
+    ProviderLookupDict,
+    AptProvider,
+    BrewProvider,
+    EnvProvider,
+)
+
+import abx
+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()
+
+    
+    # TODO: add install/load/load_or_install methods as abx.hookimpl methods
+    
+    @property
+    def admin_url(self) -> str:
+        # e.g. /admin/environment/binproviders/NpmBinProvider/   TODO
+        return "/admin/environment/binaries/"
+
+    @abx.hookimpl
+    def get_BINPROVIDERS(self):
+        return [self]
+
+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")
+
+    @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}/"
+
+    @abx.hookimpl
+    def get_BINARIES(self):
+        return [self]
+    
+
+
+apt = AptProvider()
+brew = BrewProvider()
+env = EnvProvider()